利用 sqlalchemy 实现关系表查询功能
下面的例子将完成一个通过关系表进行查询的功能，示例中的数据表均在MySQL中建立，建立过程可以使用 SQL 命令或编写 Python 适配器完成。
示例中用到的表主要有3张，一张personInfo个人信息表，一张account_store账号信息表，以及一张person_account_rel的个人信息与账号关系表。
示例中将会通过已知的人物年龄和id通过个人信息表查出个人姓名(仅为参考示例，请忽略怪异的查找逻辑 :) )，随后根据关系表得到的人物名字所对应的账号id，再根据给定的账号信息筛选出所需的账号密码结果。
完整代码如下
1 from sqlalchemy import create_engine, exc, orm 2 from sqlalchemy.ext.declarative import declarative_base 3 from sqlalchemy.sql.schema import Table, ForeignKey, Column 4 from sqlalchemy.sql.sqltypes import Integer, VARCHAR 5 from sqlalchemy.dialects.mysql.base import TINYINT 6 from sqlalchemy.orm import relationship 7 8 9 # declarative_base function will return a class which using active record pattern 10 # It will combine object opeartion and data operation automatically 11 Base = declarative_base() 12 13 # This is rel table 14 t_PersonAccount_Rel = Table('personaccount_rel', 15 Base.metadata, 16 Column('name', VARCHAR(8), ForeignKey('person_info.name')), 17 Column('account_id', Integer, ForeignKey('account_store.account_id'))) 18 19 # Create table based on Base obj 20 class PersonInfo(Base): 21 __table__ = Table('person_info', 22 Base.metadata, 23 Column('id', TINYINT(4)), 24 Column('age', Integer), 25 Column('name', VARCHAR(8), primary_key=True)) 26 27 # Need to search via person --> account 28 # So build up a relationship between person and account 29 # relationship(class_name, class, class_name) 30 AccountStore = relationship('AccountStore', 31 secondary=t_PersonAccount_Rel, 32 backref='PersonInfo') 33 34 class AccountStore(Base): 35 __table__ = Table('account_store', 36 Base.metadata, 37 Column('account_id', Integer, primary_key=True), 38 Column('items', VARCHAR(20)), 39 Column('account', VARCHAR(50)), 40 Column('password', VARCHAR(50))) 41 42 def __repr__(self): 43 return 'Items: %s\nAccount: %s\nPassword: %s' % (self.items, self.account, self.password) 44 45 class SqlalchemyActor(): 46 def __init__(self, dsn): 47 try: 48 engine = create_engine(dsn, echo=False, max_overflow=5, encoding='utf-8') 49 except ImportError: 50 raise RuntimeError 51 engine.connect() 52 53 # sessionmaker is a factory obj, generate a Session instance, reload __call__ function 54 # __call__ function will return a session class each time 55 Session = orm.sessionmaker(bind=engine) 56 # use Session() to create a class, and assign it to an attribute 57 self.session = Session() 58 # Assign costom table and engine to attribute 59 self.account = AccountStore.__table__ 60 self.engine = engine 61 # Bind engine and table 62 # Method one: assign manually one by one 63 self.account.metadata.bind = engine 64 # Method two: use reflect to map all/partly Table schema 65 #Base.metadata.reflect(engine) 66 67 class PersonInfoCriteria(): 68 """ 69 This is the criteria for PersonInfo 70 Replace None with input value 71 """ 72 def __init__(self, **kwargs): 73 self.id = None 74 self.age = None 75 self.name = None 76 self.result = None 77 78 for field, argument in kwargs.items(): 79 if str(field) == 'id': 80 self.id = argument 81 if str(field) == 'age': 82 self.age = argument 83 if str(field) == 'name': 84 self.name = argument 85 86 class PersonInfoService(): 87 """ 88 This is the service for PersonInfo 89 Generate condition and filter out expression for filter(SQL) according to criteria value 90 """ 91 92 # This function to build criteria(expression/clause) for filter(SQL) 93 # Note: PersonInfo is based on declarative_base, 94 # so PersonInfo.attr == value is an condition expression(clause) for sqlalchemy function 95 # also PersonInfo.attr.like(value) too, like function equal to "%" in SQL 96 # finally return the list of clauses 97 @staticmethod 98 def _criteria_builder(person_info_criteria): 99 clauses = [] 100 if person_info_criteria.id: 101 clauses.append(PersonInfo.id == person_info_criteria.id) 102 if person_info_criteria.age: 103 clauses.append(PersonInfo.age == person_info_criteria.age) 104 if person_info_criteria.name: 105 if '%' in person_info_criteria.name: 106 clauses.append(PersonInfo.name.like(person_info_criteria.name)) 107 else: 108 clauses.append(PersonInfo.name == person_info_criteria.name) 109 return clauses 110 111 @staticmethod 112 def find(person_info_criteria, session): 113 # Build clauses for session filter 114 clauses = PersonInfoService._criteria_builder(person_info_criteria) 115 # Query PersonInfo and filter according to clauses, use all() function to return as list 116 person_info_criteria.result = session.query(PersonInfo).filter(*clauses).all() 117 return person_info_criteria.result 118 119 class AccountStoreCriteria(): 120 def __init__(self, **kwargs): 121 self.items = None 122 self.account = None 123 self.password = None 124 self.account_id = None 125 self.person_info = None 126 self.result = None 127 128 for field, argument in kwargs.items(): 129 if field == 'items': 130 self.items = argument 131 if field == 'account': 132 self.account = argument 133 if field == 'password': 134 self.password = argument 135 if field == 'account_id': 136 self.account_id = argument 137 if field == 'person_info': 138 self.person_info = argument 139 140 class AccountStoreService(): 141 142 @staticmethod 143 def _criteria_builder(account_store_criteria): 144 clauses = [] 145 if account_store_criteria.items: 146 clauses.append(AccountStore.items == account_store_criteria.items) 147 if account_store_criteria.account: 148 if '%' in account_store_criteria.account: 149 clauses.append(AccountStore.account.like(account_store_criteria.account)) 150 else: 151 clauses.append(AccountStore.account == account_store_criteria.account) 152 if account_store_criteria.password: 153 clauses.append(AccountStore.password == account_store_criteria.password) 154 if account_store_criteria.account_id: 155 clauses.append(AccountStore.accout_id == account_store_criteria.account_id) 156 157 # person_info from PersonInfoService filter 158 # Note: pnif.AccountStore is an instrumentedList type obj 159 # sqlalchemy use instrumentedList to simulate one-to-many and many-to-many relationships 160 # sqlalchemy does not support in_ many to many relationships yet 161 # in_() function to filter out account id in range 162 # SQL: SELECT * FROM account_store WHERE account_store.account_id in (...) 163 if account_store_criteria.person_info: 164 account_ids = [] 165 for pnif in account_store_criteria.person_info: 166 for acid in pnif.AccountStore: 167 account_ids.append(acid.account_id) 168 clauses.append(AccountStore.account_id.in_(account_ids)) 169 170 return clauses 171 172 @staticmethod 173 def find(account_store_criteria, session): 174 clauses = AccountStoreService._criteria_builder(account_store_criteria) 175 account_store_criteria.result = session.query(AccountStore).filter(*clauses).all() 176 return account_store_criteria.result 177 178 if __name__ == '__main__': 179 #dsn = 'mssql+pyodbc://ItpReadOnly:@reaedonlyENC@encitp.cn.ao.ericsson.se\itp:0/ITP' 180 dsn = 'mysql+mysqldb://root:root@localhost/test_db' 181 ses = SqlalchemyActor(dsn) 182 session = ses.session 183 184 # Filter out the person information according to id and age 185 id, age = 2, 7 186 clauses = PersonInfoCriteria(id=id, age=age) 187 # re is an obj list of PersonInfo, use obj.attr to fetch value 188 person_info = PersonInfoService.find(clauses, session) 189 name = person_info[0].name 190 print('Filter out user: %s' % name) 191 192 # Filter out the account id according to name via relation table 193 items = ['WeChat', 'Qq'] 194 for it in items: 195 clauses = AccountStoreCriteria(items=it, person_info=person_info) 196 account_info = AccountStoreService.find(clauses, session) 197 for ac in account_info: 198 print(30*'-'+'\n%s' % name) 199 print(ac)