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

创建类的方式--传入一个列对象而不是一个列字符串 例如。替换代码2VS 【替换代码3

改变收集列约束的方式

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

我不知道是哪一部分的互动产生了这个错误。非常感谢任何帮助

Here is the code:

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