利用 sqlalchemy 实现关系表查询功能
下面的例子将完成一个通过关系表进行查询的功能，示例中的数据表均在MySQL中建立，建立过程可以使用 SQL 命令或编写 Python 适配器完成。
示例中用到的表主要有3张，一张personInfo个人信息表，一张account_store账号信息表，以及一张person_account_rel的个人信息与账号关系表。
示例中将会通过已知的人物年龄和id通过个人信息表查出个人姓名(仅为参考示例，请忽略怪异的查找逻辑 :) )，随后根据关系表得到的人物名字所对应的账号id，再根据给定的账号信息筛选出所需的账号密码结果。
完整代码如下
from sqlalchemy import create_engine, exc, orm
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.schema import Table, ForeignKey, Column
from sqlalchemy.sql.sqltypes import Integer, VARCHAR
from sqlalchemy.dialects.mysql.base import TINYINT
from sqlalchemy.orm import relationship # declarative_base function will return a class which using active record pattern
# It will combine object opeartion and data operation automatically
Base = declarative_base() # This is rel table
t_PersonAccount_Rel = Table('personaccount_rel',
Base.metadata,
Column('name', VARCHAR(8), ForeignKey('person_info.name')),
Column('account_id', Integer, ForeignKey('account_store.account_id'))) # Create table based on Base obj
class PersonInfo(Base):
__table__ = Table('person_info',
Base.metadata,
Column('id', TINYINT(4)),
Column('age', Integer),
Column('name', VARCHAR(8), primary_key=True)) # Need to search via person --> account
# So build up a relationship between person and account
# relationship(class_name, class, class_name)
AccountStore = relationship('AccountStore',
secondary=t_PersonAccount_Rel,
backref='PersonInfo') class AccountStore(Base):
__table__ = Table('account_store',
Base.metadata,
Column('account_id', Integer, primary_key=True),
Column('items', VARCHAR(20)),
Column('account', VARCHAR(50)),
Column('password', VARCHAR(50))) def __repr__(self):
return 'Items: %s\nAccount: %s\nPassword: %s' % (self.items, self.account, self.password) class SqlalchemyActor():
def __init__(self, dsn):
try:
engine = create_engine(dsn, echo=False, max_overflow=5, encoding='utf-8')
except ImportError:
raise RuntimeError
engine.connect() # sessionmaker is a factory obj, generate a Session instance, reload __call__ function
# __call__ function will return a session class each time
Session = orm.sessionmaker(bind=engine)
# use Session() to create a class, and assign it to an attribute
self.session = Session()
# Assign costom table and engine to attribute
self.account = AccountStore.__table__
self.engine = engine
# Bind engine and table
# Method one: assign manually one by one
self.account.metadata.bind = engine
# Method two: use reflect to map all/partly Table schema
#Base.metadata.reflect(engine) class PersonInfoCriteria():
"""
This is the criteria for PersonInfo
Replace None with input value
"""
def __init__(self, **kwargs):
self.id = None
self.age = None
self.name = None
self.result = None for field, argument in kwargs.items():
if str(field) == 'id':
self.id = argument
if str(field) == 'age':
self.age = argument
if str(field) == 'name':
self.name = argument class PersonInfoService():
"""
This is the service for PersonInfo
Generate condition and filter out expression for filter(SQL) according to criteria value
""" # This function to build criteria(expression/clause) for filter(SQL)
# Note: PersonInfo is based on declarative_base,
# so PersonInfo.attr == value is an condition expression(clause) for sqlalchemy function
# also PersonInfo.attr.like(value) too, like function equal to "%" in SQL
# finally return the list of clauses
@staticmethod
def _criteria_builder(person_info_criteria):
clauses = []
if person_info_criteria.id:
clauses.append(PersonInfo.id == person_info_criteria.id)
if person_info_criteria.age:
clauses.append(PersonInfo.age == person_info_criteria.age)
if person_info_criteria.name:
if '%' in person_info_criteria.name:
clauses.append(PersonInfo.name.like(person_info_criteria.name))
else:
clauses.append(PersonInfo.name == person_info_criteria.name)
return clauses @staticmethod
def find(person_info_criteria, session):
# Build clauses for session filter
clauses = PersonInfoService._criteria_builder(person_info_criteria)
# Query PersonInfo and filter according to clauses, use all() function to return as list
person_info_criteria.result = session.query(PersonInfo).filter(*clauses).all()
return person_info_criteria.result class AccountStoreCriteria():
def __init__(self, **kwargs):
self.items = None
self.account = None
self.password = None
self.account_id = None
self.person_info = None
self.result = None for field, argument in kwargs.items():
if field == 'items':
self.items = argument
if field == 'account':
self.account = argument
if field == 'password':
self.password = argument
if field == 'account_id':
self.account_id = argument
if field == 'person_info':
self.person_info = argument class AccountStoreService(): @staticmethod
def _criteria_builder(account_store_criteria):
clauses = []
if account_store_criteria.items:
clauses.append(AccountStore.items == account_store_criteria.items)
if account_store_criteria.account:
if '%' in account_store_criteria.account:
clauses.append(AccountStore.account.like(account_store_criteria.account))
else:
clauses.append(AccountStore.account == account_store_criteria.account)
if account_store_criteria.password:
clauses.append(AccountStore.password == account_store_criteria.password)
if account_store_criteria.account_id:
clauses.append(AccountStore.accout_id == account_store_criteria.account_id) # person_info from PersonInfoService filter
# Note: pnif.AccountStore is an instrumentedList type obj
# sqlalchemy use instrumentedList to simulate one-to-many and many-to-many relationships
# sqlalchemy does not support in_ many to many relationships yet
# in_() function to filter out account id in range
# SQL: SELECT * FROM account_store WHERE account_store.account_id in (...)
if account_store_criteria.person_info:
account_ids = []
for pnif in account_store_criteria.person_info:
for acid in pnif.AccountStore:
account_ids.append(acid.account_id)
clauses.append(AccountStore.account_id.in_(account_ids)) return clauses @staticmethod
def find(account_store_criteria, session):
clauses = AccountStoreService._criteria_builder(account_store_criteria)
account_store_criteria.result = session.query(AccountStore).filter(*clauses).all()
return account_store_criteria.result if __name__ == '__main__':
#dsn = 'mssql+pyodbc://ItpReadOnly:@reaedonlyENC@encitp.cn.ao.ericsson.se\itp:0/ITP'
dsn = 'mysql+mysqldb://root:root@localhost/test_db'
ses = SqlalchemyActor(dsn)
session = ses.session # Filter out the person information according to id and age
id, age = 2, 7
clauses = PersonInfoCriteria(id=id, age=age)
# re is an obj list of PersonInfo, use obj.attr to fetch value
person_info = PersonInfoService.find(clauses, session)
name = person_info[0].name
print('Filter out user: %s' % name) # Filter out the account id according to name via relation table
items = ['WeChat', 'Qq']
for it in items:
clauses = AccountStoreCriteria(items=it, person_info=person_info)
account_info = AccountStoreService.find(clauses, session)
for ac in account_info:
print(30*'-'+'\n%s' % name)
print(ac)
下面将分段进行解释
首先对所需的模块进行相应的导入
from sqlalchemy import create_engine, exc, orm
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.schema import Table, ForeignKey, Column
from sqlalchemy.sql.sqltypes import Integer, VARCHAR
from sqlalchemy.dialects.mysql.base import TINYINT
from sqlalchemy.orm import relationship
随后利用声明层函数产生一个声明层类，这个声明层类使用了Active Record设计模式，能够自动将对象的修改变为表格的修改
# declarative_base function will return a class which using active record pattern
# It will combine object opeartion and data operation automatically
Base = declarative_base()
接下来将进行表格建立的工作，首先建立一张关系表类对应数据库中的关系表格，此处的关系表格采用显示的ORM方式，即不基于声明层的表格类，同时定义外键信息。
# This is rel table
t_PersonAccount_Rel = Table('personaccount_rel',
Base.metadata,
Column('name', VARCHAR(8), ForeignKey('person_info.name')),
Column('account_id', Integer, ForeignKey('account_store.account_id')))
随后利用声明层建立起两张信息表，由于需要通过个人信息表与关系表两张表查找到账号的信息，因此在定义PersonInfo表类的时候，需要同时定义一个关系对象，利用relationship函数建立起表中的关系对象。
# Create table based on Base obj
class PersonInfo(Base):
__table__ = Table('person_info',
Base.metadata,
Column('id', TINYINT(4)),
Column('age', Integer),
Column('name', VARCHAR(8), primary_key=True)) # Need to search via person --> account
# So build up a relationship between person and account
# relationship(class_name, class, class_name)
AccountStore = relationship('AccountStore',
secondary=t_PersonAccount_Rel,
backref='PersonInfo') class AccountStore(Base):
__table__ = Table('account_store',
Base.metadata,
Column('account_id', Integer, primary_key=True),
Column('items', VARCHAR(20)),
Column('account', VARCHAR(50)),
Column('password', VARCHAR(50))) def __repr__(self):
return 'Items: %s\nAccount: %s\nPassword: %s' % (self.items, self.account, self.password)
最后定义一个sqlalchemy的执行器，利用基本建立步骤建立起各项对应关系
Note: 由于使用的是声明层，因此此处的metadata的绑定可以省略，这部分可参考声明层表对象的两种建立方式。
class SqlalchemyActor():
def __init__(self, dsn):
try:
engine = create_engine(dsn, echo=False, max_overflow=5, encoding='utf-8')
except ImportError:
raise RuntimeError
engine.connect() # sessionmaker is a factory obj, generate a Session instance, reload __call__ function
# __call__ function will return a session class each time
Session = orm.sessionmaker(bind=engine)
# use Session() to create a class, and assign it to an attribute
self.session = Session()
# Assign costom table and engine to attribute
self.account = AccountStore.__table__
self.engine = engine
# Bind engine and table
# Method one: assign manually one by one
self.account.metadata.bind = engine
# Method two: use reflect to map all/partly Table schema
#Base.metadata.reflect(engine)
接着，将进行信息标准类及信息筛选类的建立，利用这两个类来完成类似于SQL中的条件信息筛选。
首先是信息标准类，用于存储所需要进行处理的条件信息，根据传入的参数对筛选信息类的属性进行赋值。
class PersonInfoCriteria():
"""
This is the criteria for PersonInfo
Replace None with input value
"""
def __init__(self, **kwargs):
self.id = None
self.age = None
self.name = None
self.result = None for field, argument in kwargs.items():
if str(field) == 'id':
self.id = argument
if str(field) == 'age':
self.age = argument
if str(field) == 'name':
self.name = argument
接着建立起一个个人信息的处理服务类 PersonInfoService，在这个类中将建立起两个静态方法，首先是_criteria_builder静态方法(确切来说应该算是clauses_builder)，该静态方法会根据属性的存在，来建立起一个clauses条件语句的列表，随后再定义一个查找函数，利用query获取表格信息，再利用filter函数及前面的clause条件语句进行筛选得到最终的结果。
Note: 此处值得注意的是，由于此处的PersonInfo是一个声明层表格类，因此其==号两边的返回结果将不是bool值的True或False，而是一个等价于SQL条件的语句，一般用于filter函数中实现条件筛选。参考filter()函数的传入参数形式。
class PersonInfoService():
"""
This is the service for PersonInfo
Generate condition and filter out expression for filter(SQL) according to criteria value
""" # This function to build criteria(expression/clause) for filter(SQL)
# Note: PersonInfo is based on declarative_base,
# so PersonInfo.attr == value is an condition expression(clause) for sqlalchemy function
# also PersonInfo.attr.like(value) too, like function equal to "%" in SQL
# finally return the list of clauses
@staticmethod
def _criteria_builder(person_info_criteria):
clauses = []
if person_info_criteria.id:
clauses.append(PersonInfo.id == person_info_criteria.id)
if person_info_criteria.age:
clauses.append(PersonInfo.age == person_info_criteria.age)
if person_info_criteria.name:
if '%' in person_info_criteria.name:
clauses.append(PersonInfo.name.like(person_info_criteria.name))
else:
clauses.append(PersonInfo.name == person_info_criteria.name)
return clauses @staticmethod
def find(person_info_criteria, session):
# Build clauses for session filter
clauses = PersonInfoService._criteria_builder(person_info_criteria)
# Query PersonInfo and filter according to clauses, use all() function to return as list
person_info_criteria.result = session.query(PersonInfo).filter(*clauses).all()
return person_info_criteria.result
与前面类似，此处针对account_store类建立其标准类及服务类
Note: 此处应当注意的是，由于需要通过关系表的查询，因此需要在这个筛选标准类中多增加一项筛选标准，即传入的PersonInfo筛选结果，若传入了person_info项目，则说明需要对个人信息进行关系筛选。
class AccountStoreCriteria():
def __init__(self, **kwargs):
self.items = None
self.account = None
self.password = None
self.account_id = None
self.person_info = None
self.result = None for field, argument in kwargs.items():
if field == 'items':
self.items = argument
if field == 'account':
self.account = argument
if field == 'password':
self.password = argument
if field == 'account_id':
self.account_id = argument
if field == 'person_info':
self.person_info = argument
Note: 此处的表格服务类值得注意的是，在创建条件子句时，对于中间表的处理。由于在sqlalchemy的in_()函数尚且不支持多对多筛选，此处sqlalchemy利用instrumentedList来处理一对多或多对多的情况，在之前建立的Account_Store关系对象中，AccountStore即是instrumentList类型，可以利用instrumentList.in_(list)建立条件语句。此处利用for循环首先获取所有需要的account_id信息，生成一个列表，随后利用id列表建立等价于SQL的IN条件语句，添加到clause中。关于instrumentedList，参考*的答案。
class AccountStoreService():
@staticmethod
def _criteria_builder(account_store_criteria):
clauses = []
if account_store_criteria.items:
clauses.append(AccountStore.items == account_store_criteria.items)
if account_store_criteria.account:
if '%' in account_store_criteria.account:
clauses.append(AccountStore.account.like(account_store_criteria.account))
else:
clauses.append(AccountStore.account == account_store_criteria.account)
if account_store_criteria.password:
clauses.append(AccountStore.password == account_store_criteria.password)
if account_store_criteria.account_id:
clauses.append(AccountStore.accout_id == account_store_criteria.account_id)
# person_info from PersonInfoService filter
# Note: pnif.AccountStore is an instrumentedList type obj
# sqlalchemy use instrumentedList to simulate one-to-many and many-to-many relationships
# sqlalchemy does not support in_ many to many relationships yet
# in_() function to filter out account id in range
# SQL: SELECT * FROM account_store WHERE account_store.account_id in (...)
if account_store_criteria.person_info:
account_ids = []
for pnif in account_store_criteria.person_info:
for acid in pnif.AccountStore:
account_ids.append(acid.account_id)
clauses.append(AccountStore.account_id.in_(account_ids))
return clauses
@staticmethod
def find(account_store_criteria, session):
clauses = AccountStoreService._criteria_builder(account_store_criteria)
account_store_criteria.result = session.query(AccountStore).filter(*clauses).all()
return account_store_criteria.result
最后是执行的主程序，连接本地数据库，通过id和age筛选出name信息，随后利用关系表通过name与account_id的对应，以及所需账户类型，找到账户信息，最终显示。
if __name__ == '__main__':
dsn = 'mysql+mysqldb://root:root@localhost/test_db'
ses = SqlalchemyActor(dsn)
session = ses.session # Filter out the person information according to id and age
id, age = 2, 7
clauses = PersonInfoCriteria(id=id, age=age)
# re is an obj list of PersonInfo, use obj.attr to fetch value
person_info = PersonInfoService.find(clauses, session)
name = person_info[0].name
print('Filter out user: %s' % name) # Filter out the account id according to name via relation table
items = ['WeChat', 'Qq']
for it in items:
clauses = AccountStoreCriteria(items=it, person_info=person_info)
account_info = AccountStoreService.find(clauses, session)
for ac in account_info:
print(30*'-'+'\n%s' % name)
print(ac)
运行代码得到结果
Filter out user: LIKE
------------------------------
LIKE
Items: WeChat
Account: hereisac
Password: 12345
------------------------------
LIKE
Items: Qq
Account: re32isac
Password: 123435
从最终显示的结果可以看到，通过一系列筛选过程，得到了最终所需的账号密码信息
相关阅读
4. 声明层 ORM 访问方式