在 SQLAlchemy 中创建动态类

Question

我们有1个包含大量数据的表，DBA根据特定参数对其进行了分区。这意味着我最终得到了某种表名。以前很简单， 就像——Employee_TX, Employee_NYmodels.py

class 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):&nbsp; &nbsp; DynamicBase = declarative_base(class_registry=dict())&nbsp; &nbsp; class MyModel(DynamicBase):&nbsp; &nbsp; &nbsp; &nbsp; __tablename__ = 'Employee_{}'.format(state)&nbsp; &nbsp; &nbsp; &nbsp; name = Column...&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; state = Column...&nbsp; &nbsp; return MyModel然后从我的，我只是打电话给services.pyget_model(TX)

Answer 2

每当您想到动态构造类时，请考虑3个参数（请参阅此答案以获取演示，以及更一般的文档）。type()在你的例子中，这只是一个构造类并保留对它们的引用的问题，以便你以后可以再次访问它们。下面是一个示例：from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.engine import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerBase = declarative_base()# this produces the set of common attributes that each class should havedef attribute_factory():&nbsp; &nbsp; return dict(&nbsp; &nbsp; &nbsp; &nbsp; id=Column(Integer, primary_key=True),&nbsp; &nbsp; &nbsp; &nbsp; name=Column(String, nullable=False),&nbsp; &nbsp; &nbsp; &nbsp; state=Column(String, nullable=False),&nbsp; &nbsp; &nbsp; &nbsp; CLASS_VAR=12345678,&nbsp; &nbsp; )states = ["CA", "TX", "NY"]# here we map the state abbreviation to the generated model, notice the templated# class and table namesmodel_map = {&nbsp; &nbsp; state: type(&nbsp; &nbsp; &nbsp; &nbsp; f"Employee_{state}",&nbsp; &nbsp; &nbsp; &nbsp; (Base,),&nbsp; &nbsp; &nbsp; &nbsp; dict(**attribute_factory(), __tablename__=f"Employee_{state}"),&nbsp; &nbsp; )&nbsp; &nbsp; for state in states}engine = create_engine("sqlite:///", echo=True)Session = sessionmaker(bind=engine)Base.metadata.create_all(engine)if __name__ == "__main__":&nbsp; &nbsp; # inserts work&nbsp; &nbsp; s = Session()&nbsp; &nbsp; for state, model in model_map.items():&nbsp; &nbsp; &nbsp; &nbsp; s.add(model(name="something", state=state))&nbsp; &nbsp; s.commit()&nbsp; &nbsp; s.close()&nbsp; &nbsp; # queries work&nbsp; &nbsp; s = Session()&nbsp; &nbsp; for state, model in model_map.items():&nbsp; &nbsp; &nbsp; &nbsp; inst = s.query(model).first()&nbsp; &nbsp; &nbsp; &nbsp; print(inst.state, inst.CLASS_VAR)