在 SQLAlchemy 中创建动态类

Question

我们有 1 个包含大量数据的表，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

通过创建自定义函数来获得它，例如 -

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)