首頁猿問在 SQLAlchemy 中創建動態類

在 SQLAlchemy 中創建動態類

Python

慕尼黑5688855 2022-08-25 15:51:46

我們有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。

查看完整描述

2 回答

慕田峪4524236

TA貢獻1875條經驗獲得超5個贊

通過創建自定義函數（如 -

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)

反對回復 2022-08-25

BIG陽

TA貢獻1859條經驗獲得超6個贊

每當您想到動態構造類時，請考慮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)

反對回復 2022-08-25

2 回答
0 關注
410 瀏覽

關注

添加回答

舉報

0/150

提交

取消

亚洲在线久爱草,狠狠天天香蕉网,天天搞日日干久草,伊人亚洲日本欧美

熱搜

最近搜索清空

在 SQLAlchemy 中創建動態類

在 SQLAlchemy 中創建動態類

2 回答

添加回答