How to validate row if it's compliant before pushing it to pgsql?

Viewed 514

Say table structure is like:

CREATE TABLE foo (
    id INTEGER  NOT NULL,
    enter_time TIMESTAMP NOT NULL,
    comment TEXT
);   

Now in python, say I get data like this:

foo_user = {"id": 123, "enter_time": None, "comment": ''}

How can I manually validate this data before sending this to pgsql?

Is there any library which already do this by pulling schema information from pgsql and doing validation on that?

6 Answers

The first step is to retrieve the expected data type from the DB itself, this can be accomplished, as suggested by @gaurav) using a:

SELECT column_name, data_type FROM information_schema.columns where ...

This gives to you the type schema, this can be used as 'validation schema'

simple example - diy

Here a simple example limited to check if input data can be typed (it will be casted by postgres in the same manner, maybe)

from datetime import datetime
schema = {"id": "integer", "enter_time": "TIMESTAMP", "comment": 'text'}

def is_valid(v, validator):
    # dummy validator, we try to apply a constructor
    # returns only True/False, 
    # If False... we don't know why!
    # here you can use a RE for check if input is syntactically correct
    try:
        validator(v)
    except:
        return False
    else:
        return True
    


# type to validator
binding = {'integer':int,
           'TIMESTAMP': datetime.fromtimestamp,
           'text':str}

#validation schemaa
val_schema = {k:binding[v] for k,v in schema.items()}



foo_user = {"id": 123, "enter_time": None, "comment": ''}

for k,v in foo_user.items():
    print (k, ':', is_valid(v, val_schema[k]))


# 
# id : True
# enter_time : False
# comment : True

better approach

For the second step, there are specialized validation libraries, they can do typing, clipping, and schema validation (like 2 fields password that must be identical) and a lot of useful stuff.

I've worked a lot with voluptuous but many choices are available up to now, do a good survey before adopt one of that library in your lib stack.

This query returns "mytable1" column names,their types and default values

 SELECT attname 
      ,pg_catalog.format_type(a.atttypid, a.atttypmod) as Datatype 
      ,coalesce(d.adsrc,'') AS default_value
  FROM pg_catalog.pg_attribute a 
  LEFT JOIN pg_catalog.pg_attrdef d ON (a.attrelid, a.attnum)=(d.adrelid,d.adnum)
 WHERE a.attnum > 0 AND NOT a.attisdropped
   AND a.attrelid = ( SELECT c.oid
                       FROM pg_catalog.pg_class c 
                       LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace 
                      WHERE c.relname = 'mytable1' AND pg_catalog.pg_table_is_visible(c.oid))
              

If you are willing to add ORM there are a couple of ways to retrieve your schema.

Automap from SQLAlchemy: https://docs.sqlalchemy.org/en/14/orm/extensions/automap.html

It will map your schema to classes and then you will be able to use this mapping to insert your data.

There is also this package https://pypi.org/project/sqlacodegen/ which generates the tables as classes in the codebase.

If you are looking to extract info from DB directly regarding columns PostgreSQL information_schema columns table holds this data. For example:

SELECT  
  column_name,
  ordinal_position,
  column_default,
  is_generated,
  is_nullable,
  data_type
FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'test';

Or for a complete schema

SELECT  
   column_name,
   ordinal_position,
   column_default,
   is_generated,
   is_nullable,
   data_type
FROM INFORMATION_SCHEMA.COLUMNS WHERE  table_schema='test';

if I am correctly understand this then you want schema of table so you match with current input before inserting to database. if yes then

cur.execute("SELECT column_name, data_type FROM information_schema.columns WHERE table_name = 'foo';")
print(dict(cur.fetchall()))
#{"id": "integer", "enter_time": "TIMESTAMP", "comment": 'text'}

For that you can use ORMs. SQLAlchemy is used by many.

Using pydantic, you could validate inputs as:

from datetime import datetime
from pydantic import BaseModel, validator

class FooModel(BaseModel):
    id: int
    enter_time: datetime
    comment: str

    @validator('id')
    def id_cannot_be_none(cls, v):
        if v is None:
            raise ValueError('id can not be None')
        return v
    
    @validator('enter_time')
    def enter_time_cannot_be_none(cls, v):
        if v is None:
            raise ValueError('enter_time can not be None')
        return v

# validate inputs 
foo_user = {"id": 123, "enter_time": None, "comment": ''}

validate_foo_user = FooModel(**foo_user)

NB: we might need timestamp type hint instead of datetime

Related