在SQLAlchemy中创建动态类

Question

我们有一个包含大量数据的表，DBA根据特定参数对其进行了分区。这意味着我最终得到了Employee_TX, Employee_NY类的表名。前面的models.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操作时，它也找不到生成的类

这是一个flask应用程序，PostgreSQL作为后端，sqlalchemy用作ORM

Answer 1

每当您想到动态构造类时，都会想到带有3个参数的type()（请参见this answer的演示，以及更一般的the docs）

在您的情况下，只需构造类并保留对它们的引用，以便以后可以再次访问它们

下面是一个例子：

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)

Answer 2

通过创建一个自定义函数，如-

def get_model(state):
    DynamicBase = declarative_base(class_registry=dict())

    class MyModel(DynamicBase):
        __tablename__ = 'Employee_{}'.format(state)

        name = Column...         
        state = Column...

return MyModel

然后从我的services.py开始，我用get_model(TX)调用