利用 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:@[email protected]\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)

下面將分段進行解釋
首先對所需的模塊進行相應的導入

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，參考stackoverflow的答案。

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

從最終顯示的結果可以看到，通過一系列篩選過程，得到了最終所需的賬號密碼信息

相關閱讀

1. ORM 與 sqlalchemy 模塊

2. sqlalchemy 的基本使用

Python與數據庫[2] -> 關系對象映射/ORM -> 利用 sqlalchemy 實現關系表查詢功能