Python与数据库[2] -> 关系对象映射/ORM -> 利用 sqlalchemy 实现关系表查询功能

利用 sqlalchemy 实现关系表查询功能

示例中用到的表主要有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:[email protected]/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:[email protected]/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

从最终显示的结果可以看到，通过一系列筛选过程，得到了最终所需的账号密码信息

相关阅读

2. sqlalchemy 的基本使用

原文地址：https://www.cnblogs.com/stacklike/p/8186253.html