Создание динамических c классов в SQLAlchemy - PullRequest
Новая
       2

Создание динамических c классов в SQLAlchemy

1 голос
/ 10 марта 2020

У нас есть 1 таблица с большим объемом данных, и администратор БД разбил ее на основе определенного параметра. Это означает, что я получил Employee_TX, Employee_NY вид имен таблиц. Ранее models.py было просто, как в - 

class Employee(Base):
    __tablename__ = 'Employee'
    name = Column...
    state = Column...

Теперь я не хочу создавать 50 новых классов для вновь разбитых таблиц, так как в любом случае мои столбцы одинаковы.

* 1007 Есть ли шаблон, в котором я могу создать отдельный класс и затем использовать его в запросе динамически? session.query(<Tablename>).filter().all()

Возможно, я ищу какой-то шаблон фабрики или что-то в этом роде.

До сих пор я пробовал запустить al oop как 

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

Это * Приложение 1024 * с PostgreSQL в качестве бэкэнда и sqlalchemy используется как ORM

Ответы [ 2 ]

2 голосов
/ 10 марта 2020

Всякий раз, когда вы думаете о динамически создаваемых классах, думайте о type() с 3 аргументами (см. этот ответ для демонстрации и документы в более общем плане).

В вашем случае это просто вопрос создания классов и сохранения ссылки на них, чтобы вы могли получить к ним доступ позже.

Вот пример: 

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)
1 голос
/ 10 марта 2020

Получил это, создав пользовательскую функцию, такую ​​как - 

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)

...