在 SQLAlchemy 中创建动态类

Question

我们有1个包含大量数据的表，DBA根据特定参数对其进行了分区。这意味着我最终得到了某种表名。以前很简单， 就像——Employee_TX, Employee_NYmodels.pyclass Employee(Base):    __tablename__ = 'Employee'    name = Column...    state = Column...现在，我不想为新分区的表创建50个新类，因为无论如何，我的列是相同的。有没有一种模式，我可以创建一个类，然后在查询中动态使用它？session.query(<Tablename>).filter().all()也许某种工厂模式或其他东西是我正在寻找的。到目前为止，我已经尝试通过运行循环作为for state in ['CA', 'TX', 'NY']:    class Employee(Base):        __qualname__ = __tablename__ = 'Employee_{}'.format(state)        name = Column...        state = Column...但这不起作用，我得到一个警告 -SAWarning: This declarative base already contains a class with the same class name and module name as app_models.employee, and will be replaced in the string-lookup table.此外，当我这样做时，它找不到生成的类from app_models import Employee_TX这是一个烧瓶应用程序，以PostgreSQL作为后端，sqlalchemy用作ORM。

Answer 1

通过创建自定义函数（如 -

def get_model(state):

    DynamicBase = declarative_base(class_registry=dict())

    class MyModel(DynamicBase):

        __tablename__ = 'Employee_{}'.format(state)

        name = Column...         

        state = Column...

    return MyModel

然后从我的，我只是打电话给services.pyget_model(TX)

Answer 2

每当您想到动态构造类时，请考虑3个参数（请参阅此答案以获取演示，以及更一般的文档）。type()

在你的例子中，这只是一个构造类并保留对它们的引用的问题，以便你以后可以再次访问它们。

下面是一个示例：

from sqlalchemy import Column, Integer, String

from sqlalchemy.engine import create_engine

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import sessionmaker

Base = declarative_base()

# this produces the set of common attributes that each class should have

def attribute_factory():

    return dict(

        id=Column(Integer, primary_key=True),

        name=Column(String, nullable=False),

        state=Column(String, nullable=False),

        CLASS_VAR=12345678,

    )

states = ["CA", "TX", "NY"]

# here we map the state abbreviation to the generated model, notice the templated

# class and table names

model_map = {

    state: type(

        f"Employee_{state}",

        (Base,),

        dict(**attribute_factory(), __tablename__=f"Employee_{state}"),

    )

    for state in states

}

engine = create_engine("sqlite:///", echo=True)

Session = sessionmaker(bind=engine)

Base.metadata.create_all(engine)

if __name__ == "__main__":

    # inserts work

    s = Session()

    for state, model in model_map.items():

        s.add(model(name="something", state=state))

    s.commit()

    s.close()

    # queries work

    s = Session()

    for state, model in model_map.items():

        inst = s.query(model).first()

        print(inst.state, inst.CLASS_VAR)