如何使用 SQLAlchemy Postgres ORM 的声明性基础动态创建具有列名和字典约束的表？

Posted 2023-03-16

【中文标题】如何使用 SQLAlchemy Postgres ORM 的声明性基础动态创建具有列名和字典约束的表？

【英文标题】：How to Dynamically Create Tables With Column Names and Constraints From Dictionary Using SQLAlchemy Postgres ORM's Declarative Base?

【发布时间】：2021-12-18 09:13:53

【问题描述】：

设置：Postgres13、Python 3.7、SQLAlchemy 1.4

我的问题是关于动态创建类而不是依赖于models.py 的内容。我有一个 schema.json 文件，其中包含许多表的元数据。列数、列名、列约束因表而异，事先不知道。

解析 JSON 并将其结果映射到 ORM Postgres 方言（例如：'column_name1': 'bigint' 变为 'column_name1 = Column(BigInt)'）。这将创建一个字典，其中包含 表名、列名和列约束。由于所有表格都通过了增强基数，因此它们会自动通过 接收一个 PK id 字段。

然后我将此字典传递给create_class 函数，该函数使用此数据动态创建表 并将这些新表提交到数据库。

挑战在于，当我运行代码时，确实会创建表，但只有一列 - PK id 它自动收到。所有其他列都将被忽略。

我怀疑我在制造这个错误的方式 我正在调用 Session 或 Base 或以我传递列约束的方式。我不确定如何向 ORM 表明我正在传递 Column 和 Constraint 对象。

我尝试过更改以下内容：

类的创建方式——传入一个 Column 对象而不是一个 Column 字符串 例如：constraint_dict[k] = f'= Column(v)' VS constraint_dict[k] = f'= Column(v)'

改变收集列约束的方式

以不同的方式调用Base 和create。我尝试在下面create_class 中的注释行中显示这些变化。

我无法确定是哪些交互导致了此错误。非常感谢任何帮助！

代码如下：

schema.json 示例

  "groupings": 
    "imaging": 
      "owner":  "type": "uuid", "required": true, "index": true ,
      "tags":  "type": "text", "index": true 
      "filename":  "type": "text" ,
    ,

    "user": 
      "email":  "type": "text", "required": true, "unique": true ,
      "name":  "type": "text" ,
      "role": 
        "type": "text",
        "required": true,
        "values": [
          "admin",
          "customer",
        ],
        "index": true
      ,
      "date_last_logged":  "type": "timestamptz" 
    
  ,

  "auths": 
    "boilerplate": 
      "owner": ["read", "update", "delete"],
      "org_account": [],
      "customer": ["create", "read", "update", "delete"]
     ,

      "loggers": 
      "owner": [],
      "customer": []
    
  

base.py

from sqlalchemy import Column, create_engine, Integer, MetaData
from sqlalchemy.orm import declared_attr, declarative_base, scoped_session, sessionmaker

engine = create_engine('postgresql://user:pass@localhost:5432/dev', echo=True)

db_session = scoped_session(
    sessionmaker(
        bind=engine,
        autocommit=False,
        autoflush=False
    )
)
# Augment the base class by using the cls argument of the declarative_base() function so all classes derived
# from Base will have a table name derived from the class name and an id primary key column.
class Base:
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

    id = Column(Integer, primary_key=True)

metadata_obj = MetaData(schema='collect')
Base = declarative_base(cls=Base, metadata=metadata_obj)

models.py

from base import Base
from sqlalchemy import Column, DateTime, Integer, Text
from sqlalchemy.dialects.postgresql import UUID
import uuid


class NumLimit(Base):

    org = Column(UUID(as_uuid=True), default=uuid.uuid4, unique=True)
    limits = Column(Integer)
    limits_rate = Column(Integer)
    rate_use = Column(Integer)

    def __init__(self, org, limits, allowance_rate, usage, last_usage):
        super().__init__()
        self.org = org
        self.limits = limits
        self.limits_rate = limits_rate
        self.rate_use = rate_use

create_tables.py（我知道这个很乱！只是试图显示所有尝试的变体......）

def convert_snake_to_camel(name):
    return ''.join(x.capitalize() or '_' for x in name.split('_'))


def create_class(table_data):
    constraint_dict = '__tablename__': 'TableClass'
    table_class_name = ''
    column_dict = 

    for k, v in table_data.items():

        # Retrieve table, alter the case, store it for later use
        if 'table' in k:
            constraint_dict['__tablename__'] = v
            table_class_name += convert_snake_to_camel(v)

        # Retrieve the rest of the values which are the column names and constraints, ex: 'org = Column(UUID(as_uuid=True), default=uuid.uuid4, unique=True)'
        else:
            constraint_dict[k] = f'= Column(v)'
            column_dict[k] = v

    # When type is called with 3 arguments it produces a new class object, so we use it here to create the table Class
    table_cls = type(table_class_name, (Base,), constraint_dict)

    # Call ORM's 'Table' on the Class
    # table_class = Table(table_cls) # Error "TypeError: Table() takes at least two positional-only arguments 'name' and 'metadata'"

    # db_session.add(table_cls) # Error "sqlalchemy.orm.exc.UnmappedInstanceError: Class 'sqlalchemy.orm.decl_api.DeclarativeMeta'
                                # is not mapped; was a class (__main__.Metadata) supplied where an instance was required?"

    # table_class = Table(
    #    table_class_name,
    #    Base.metadata,
    #    constraint_dict) # Error "sqlalchemy.orm.exc.UnmappedInstanceError: Class 'sqlalchemy.orm.decl_api.DeclarativeMeta'
                          # is not mapped; was a class (__main__.Metadata) supplied where an instance was required?"

    # table_class = Table(
    #    table_class_name,
    #    Base.metadata,
    #    column_dict)  
    #    table_class.create(bind=engine, checkfirst=True) # sqlalchemy.exc.ArgumentError: 'SchemaItem' object, such as a 'Column' or a 'Constraint' expected, got 'limits': 'Integer'
 
    # table_class = Table(
    #    table_class_name,
    #    Base.metadata,
    #    **column_dict) # TypeError: Additional arguments should be named <dialectname>_<argument>, got 'limits'

    # Base.metadata.create_all(bind=engine, checkfirst=True)
    # table_class.create(bind=engine, checkfirst=True)
    
    new_row_vals = table_cls(**column_dict)
    db_session.add(new_row_vals)  # sqlalchemy.exc.ArgumentError: 'SchemaItem' object, such as a 'Column' or a 'Constraint' expected, got 'limits': 'Integer'

    db_session.commit()
    db_session.close()

【问题讨论】：

我缺少能够回答这个问题的是您的输入数据是什么样的。你能添加一个示例 schema.json 吗？

@JesseBakker 我刚刚在问题的顶部添加了一个 JSON 示例，谢谢！

也在 GitHub 上回答 here

【参考方案1】：

我为您创建了一个独立的示例。这应该为您提供自己构建它的基本构建块。它包括类型映射，将类型字符串映射到 sqlalchemy 类型和参数映射，将非 sqlalchemy 参数映射到它们的 sqlalchemy 对应项（required: True 在 sqlalchemy 中是 nullable: False）。 此方法使用metadata 定义表，然后将它们转换为声明性映射，如Using a Hybrid Approach with __table__ 中使用python type() 函数所述。然后将这些生成的类导出到模块范围的globals()。

并非您提供的schema.json 中的所有内容都受支持，但这应该会给您一个很好的起点。

from sqlalchemy import Column, Integer, Table, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import declarative_base


def convert_snake_to_camel(name):
    return "".join(part.capitalize() for part in name.split("_"))


data = 
    "groupings": 
        "imaging": 
            "id": "type": "integer", "primary_key": True,
            "owner": "type": "uuid", "required": True, "index": True,
            "tags": "type": "text", "index": True,
            "filename": "type": "text",
        ,
        "user": 
            "id": "type": "integer", "primary_key": True,
            "email": "type": "text", "required": True, "unique": True,
            "name": "type": "text",
            "role": 
                "type": "text",
                "required": True,
                "index": True,
            ,
        ,
    ,

Base = declarative_base()

typemap = 
    "uuid": UUID,
    "text": Text,
    "integer": Integer,


argumentmap = 
    "required": lambda value: ("nullable", not value),


for tablename, columns in data["groupings"].items():
    column_definitions = []
    for colname, parameters in columns.items():
        type_ = typemap[parameters.pop("type")]
        params = 
        for name, value in parameters.items():
            try:
                name, value = argumentmap[name](value)
            except KeyError:
                pass
            finally:
                params[name] = value
        column_definitions.append(Column(colname, type_(), **params))

    # Create table in metadata
    table = Table(tablename, Base.metadata, *column_definitions)

    classname = convert_snake_to_camel(tablename)
    # Dynamically create a python class with definition
    # class classname:
    #     __table__ = table
    class_ = type(classname, (Base,), "__table__": table)

    # Add the class to the module namespace
    globals()[class_.__name__] = class_

【讨论】：

感谢您创建此示例，这是一个很好的垫脚石！