如何在python sqlalchemy中拥有select()的公共模块

Question

我的表有3个来自SQLAlchemy的模型类Table1 Table2 Table3

''

from sqlalchemy import create_engine , text , select, MetaData, Table  ,func , Column , String , Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
import sqlalchemy
from settings import DATABASE_URI
engine=create_engine(DATABASE_URI)
Base = declarative_base()
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
metadata = MetaData(bind=None)
session = Session()
class Table1(Base):
__tablename__ = 'table1'
id = Column(u'id', Integer(), primary_key=True)
name1 = Column(u'name1', String(50))
class Table2(Base):
__tablename__ = 'table2'
id = Column(u'id', Integer(), primary_key=True)
name2 = Column(u'name2', String(50))
class Table3(Base):
__tablename__ = 'table3'
id = Column(u'id', Integer(), primary_key=True)
name3 = Column(u'name3', String(50))

class connectionTest():
def wrapper_connection(self,table,column,value):
#SELECT column FROM table1 WHERE column = value
query = session.query(table)
q = query.filter_by(column = value)
session.execute(q)

def main():
ct = connectionTest()
t1 = Table1()
t2 = Table2()
t3 = Table3()
ct.wrapper_connection(t1,t1.name1, "Table1_Value_Information")
ct.wrapper_connection(t2,t2.name2, "Table2_Value_Information")
ct.wrapper_connection(t3,t3.name3, "Table3_Value_Information")
if __name__ == '__main__':
main()

''

我希望包装器连接能够处理具有不同列的所有3个不同的表。基本上是为了使其尽可能通用，以便通过SQLAlchemy ORM或Core库处理与(#SELECT column FROM table1 WHERE column=value(子句相关的任何条件。

我面临的问题是这样的。'q=query.filter_by(column=value('

其中我试图传递来自函数属性t1.name1 的列信息

ct.wrapper_connection(t1，t1.name1，"Table1_Value_Information"(

我面临的错误：

Traceback(最后一次调用(：文件"；C： \用户<用户名>\AppData\Local\Programs\Python37\lib\site packages\sqlalchemy\orm\base.py"；，第406行，在_entity_descriptor中返回getattr(实体，密钥(

AttributeError：类型对象"Table1"没有属性"column">

Answer 1

问题中的代码需要三个更改：

• 在wrapper_connection方法中，使用Query.filter而不是Query.filter_by，因为它将直接接受列对象
• 在声明模型类之前，不要调用Base.metadata.create_all()
• t1 = Table1()创建Table1类的新实例-一行。您希望针对表进行查询，因此请直接使用模型类

class connectionTest:
def wrapper_connection(self, table, column, value):
# SELECT column FROM table1 WHERE column = value
query = session.query(table)
# We have the column object, so use filter
# instead of filter_by
q = query.filter(column == value)
session.execute(q)

# Create the tables after the model classes have been declared.
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

def main():
ct = connectionTest()
# Use the model classes directly.
ct.wrapper_connection(Table1, Table1.name1, "Table1_Value_Information")
ct.wrapper_connection(Table2, Table2.name2, "Table2_Value_Information")
ct.wrapper_connection(Table3, Table3.name3, "Table3_Value_Information")