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

Question

设置：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()

Answer 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_