如何在将行推送到 pgsql 之前验证行是否符合要求？

Question

说表结构是这样的：

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

现在在 python 中，假设我得到这样的数据：

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

如何在发送到 pgsql 之前手动验证这些数据？

是否有任何库已经通过从 pgsql 中提取架构信息并对其进行验证来做到这一点？

Answer 1

第一步是从数据库本身检索预期的数据类型，这可以通过@gaurav 的建议来完成）使用：

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

这给了你类型模式，这可以用作'验证模式'

简单例子-diy

这里是一个简单的例子，仅限于检查输入数据是否可以输入（它可能会被 postgres 以相同的方式强制转换）

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

更好的方法

对于第二步，有专门的验证库，它们可以进行打字、剪辑和模式验证（例如必须相同的 2 个字段密码）以及许多有用的东西。

我与voluptuous 合作了很多，但到目前为止many choices 仍然可用，在您的库堆栈中采用该库之前，请先做好调查。

Answer 2

此查询返回“mytable1”列名、它们的类型和默认值

 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))
              

Answer 3

如果您愿意添加 ORM，有几种方法可以检索您的架构。

来自 SQLAlchemy 的自动映射： https://docs.sqlalchemy.org/en/14/orm/extensions/automap.html

它将您的架构映射到类，然后您将能够使用此映射插入数据。

还有这个包https://pypi.org/project/sqlacodegen/ 将表格生成为代码库中的类。

如果您希望直接从数据库中提取有关列的信息 PostgreSQL information_schema 列表包含此数据。 例如：

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

或者对于一个完整的架构

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

Answer 4

如果我正确理解这一点，那么您需要表的架构，以便在插入数据库之前与当前输入匹配。如果是那么

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'}

Answer 5

为此，您可以使用 ORM。 SQLAlchemy 被许多人使用。

Answer 6

使用pydantic，您可以将输入验证为：

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)

注意：我们可能需要时间戳类型提示而不是日期时间