Jusene's Blog

Sqlalchemy ORM框架(二)

python
字数统计: 2.3k阅读时长: 12 min
2018/10/07 Share

建立关系

relationship函数是sqlalchemy对关系之间提供一种便利调用方式，back_populates参数则对关系提供反向引用的声明。

1
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
2
from sqlalchemy.orm import sessionmaker, relationship
3
from sqlalchemy.ext.declarative import declarative_base
4
5
Model = declarative_base()
6
7
class User(Model):
8
    __tablename__ = 'users'
9
    id = Column(Integer, primary_key=True, autoincrement=True)
10
    name = Column(String(64))
11
    fullname = Column(String(128))
12
    password = Column(String(128))
13
    addresses = relationship("Address", order_by=Address.id, back_populates='user')
14
15
    def __repr__(self):
16
        return "<User(name={}, fullname={})".format(self.name, self.fullname)
17
18
class Address(Model):
19
    __tablename__ = 'addresses'
20
    id = Column(Integer, primary_key = True, autoincrement=True)
21
    email_address = Column(String(64), nullable=False)
22
    user_id = Column(Integer, ForeignKey('users.id'))
23
    user = relationship('User', back_populates='addresses')
24
25
    def __repr__(self):
26
        return "<Address(email_address={})>".format(self.email_address)
27
28
29
engine = create_engine("mysql+pymysql://root:123456@localhost:3306/mytest?charset=utf8mb4", echo=True)
30
Model.metadata.create_all(engine)
31
32
Session = sessionmaker(bind=engine)
33
session = Session()
34
35
jusene = User(name='jusene', fullname='Jusene Ben', password='1q2w3ee')
36
jusene.addresses
37
[]
38
39
jusene.addresses = [
40
    Address(email_address='jusene@123.com'),
41
    Address(email_address='jusene@163.com')
42
]
43
44
jusene.addresses[1].user
45
<User(name='jusene'), fullname='Jusene Ben'>
46
47
session.add(jusene)
48
session.commit()

连表查询

1
# SQL: SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password, addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id FROM users, addresses WHERE users.id = addresses.user_id AND addresses.email_address = 'jusene@123.com'
2
for u, a in  session.query(User, Address).filter(User.id == Address.user_id).filter(Address.email_address == 'jusene@123.com').all():
3
    print(u)
4
    print(a)
5
6
# 内连接
7
# SQL: SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password FROM users INNER JOIN addresses ON users.id = addresses.user_id WHERE addresses.email_address = 'jusene@123.com'
8
session.query(User).join(Address).filter(Address.email_address == 'jusene@123.com').all()
9
10
# 左外连接
11
# SQL: SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password FROM users LEFT OUTER JOIN addresses ON users.id = addresses.user_id WHERE addresses.email_address = 'jusene@123.com'
12
session.query(User).outerjoin(Address).filter(Address.email_address == 'jusene@123.com').all()

使用别名

1
# SQL: SELECT users.name AS users_name, addresses_1.email_address AS addresses_1_email_address, addresses_2.email_address AS addresses_2_email_address FROM users INNER JOIN addresses AS addresses_1 ON users.id = addresses_1.user_id INNER JOIN addresses AS addresses_2 ON users.id = addresses_2.user_id WHERE addresses_1.email_address = 'jusene@163.com' AND addresses_2.email_address = 'jusene@123.com'
2
3
from sqlalchemy.orm import aliased
4
5
ad1 = aliased(Address)
6
ad2 = aliased(Address)
7
8
for username, email1, email2 in \
9
    session.query(User.name, ad1.email_address, ad2.email_address). \
10
    join(ad1, User.addresses). \
11
    join(ad2, User.addresses). \
12
    filter(ad1.email_address == 'jusene@123.com'). \
13
    filter(ad2.email_address == 'jusene@163.com'):
14
15
    print(username, email1, email2)

使用子查询

1
from sqlalchemy.sql import func
2
3
# SQL: SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password, anon_1.address_count AS anon_1_address_count FROM users LEFT OUTER JOIN (SELECT addresses.user_id AS user_id, count('*') AS address_count FROM addresses GROUP BY addresses.user_id) AS anon_1 ON users.id = anon_1.user_id ORDER BY users.id
4
sub = session.query(Address.user_id, func.count('*').label('address_count')).group_by(Address.user_id).subquery()
5
for u, count in session.query(User, sub.c.address_count).outerjoin(sub, User.id == sub.c.user_id).order_by(User.id):
6
    print(u, count)

将子查询作为入口

1
# SQL: SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password, anon_1.id AS anon_1_id, anon_1.email_address AS anon_1_email_address, anon_1.user_id AS anon_1_user_id FROM users INNER JOIN (SELECT addresses.id AS id, addresses.email_address AS email_address, addresses.user_id AS user_id FROM addresses WHERE addresses.email_address = 'jusene@163.com' AS anon_1 ON users.id = anon_1.user_id
2
stmt = session.query(Address).filter(Address.email_address == 'jusene@163.com').subquery()
3
ad = aliased(Address, stmt)
4
for user, address in session.query(User, ad).join(ad, User.addresses):
5
    print(user)
6
    print(address)

使用exists

  • exists

    1
    from sqlalchemy.sql import exists
    2
    3
    # SQL: SELECT users.name AS users_name FROM users WHERE EXISTS (SELECT * FROM addresses WHERE addresses.user_id = users.id)
    4
    stmt = exists().where(Address.user_id == User.id)
    5
    for name, in session.query(User.name).filter(stmt):
    6
        print(name)

  • any(relationship)

    1
    # SQL: SELECT users.name AS users_name FROM users WHERE EXISTS (SELECT 1 FROM addresses WHERE users.id = addresses.user_id)
    2
    for name, in session.query(User.name).filter(User.addresses.any()):
    3
        print(name)
1
# SQL: SELECT users.name AS users_name FROM users WHERE EXISTS (SELECT 1 FROM addresses WHERE users.id = addresses.user_id AND addresses.email_address LIKE "%163%")
2
for name, in session.query(User.name).filter(User.addresses.any(Address.email_address.like('%163%'))):
3
    print(name)
  • has
    1
    # SQL: SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id FROM addresses WHERE NOT (EXISTS (SELECT 1 FROM users WHERE users.id = addresses.user_id AND users.name = 'jusene')
    2
    session.query(Address).filter(~Address.user.has(User.name == 'jusene')).all()

表关系

一对多

表A(一)与表B(多),表A中的一条记录对应表B中的多条记录之间要有关系，就必须要有外键，把表A的主键添加到表B里面，充当表B的外键，一对多的实现，在多的一方表里面添加外键。

1
class Parent(Model):
2
    __tablename__ = 'parent'
3
    id = Column(Integer, primary_key=True)
4
    name = Column(String(64), nullable=False)
5
    children = relationship('Child')
6
7
    def __repr__(self):
8
        return "<Parent(name={})>".format(self.name)
9
10
class Child(Model):
11
    __tablename__ = 'child'
12
    id = Column(Integer, primary_key=True)
13
    email = Column(String(128), nullable=False)
14
    parent_id = Column(Integer, ForeignKey('parent.id'))
15
16
    def __repr__(self):
17
        return "<Child(email={})>".format(self.email)
18
19
jusene = Parent(name='jusene')
20
jusene.children
21
[]
22
jusene.children = [Child(email='jusene@163.com'), Child(email='jusene@qq.com')]
23
session.add(jusene)
24
session.commit()
25
26
for row in session.query(Parent):
27
    print(row.children[1].email)

多对一

在一对多的关系中建立双向关系，这样的话在对方看来就是多对一的关系，在子类表附加一个relationship方法，并且在双方的relationship方法中使用back_populates方法。

1
class Parent(Model):
2
    __tablename__ = 'parent'
3
    id = Column(Integer, primary_key=True)
4
    name = Column(String(64), nullable=False)
5
    children = relationship('Child',back_populates='parent')
6
7
    def __repr__(self):
8
        return "<Parent(name={})>".format(self.name)
9
10
class Child(Model):
11
    __tablename__ = 'child'
12
    id = Column(Integer, primary_key=True)
13
    email = Column(String(128), nullable=False)
14
    parent_id = Column(Integer, ForeignKey('parent.id'))
15
    parent = relationship('Parent', back_populates='children')
16
17
    def __repr__(self):
18
        return "<Child(email={})>".format(self.email)
19
20
for row in session.query(Child).filter(Child.email == 'jusene@qq.com'):
21
    print(row.parent.name)

这样的话子表将在多对一的关系中获得父表的属性

或者在单一的relationship方法中使用backref参数代替back_populates参数:

1
class Parent(Model):
2
    __tablename__ = 'parent'
3
    id = Column(Integer, primary_key=True)
4
    name = Column(String(64), nullable=False)
5
    children = relationship('Child',backref='parent')
6
7
    def __repr__(self):
8
        return "<Parent(name={})>".format(self.name)
9
10
class Child(Model):
11
    __tablename__ = 'child'
12
    id = Column(Integer, primary_key=True)
13
    email = Column(String(128), nullable=False)
14
    parent_id = Column(Integer, ForeignKey('parent.id'))
15
16
    def __repr__(self):
17
        return "<Child(email={})>".format(self.email)

一对一

在一对一关系中，A 表中的一行最多只能匹配于 B 表中的一行，反之亦然。如果相关列都是主键或都具有唯一约束，则可以创建一对一关系。

一对一是两张表之间本质的双向关系，要做到这一点，需要在一对多的关系基础上的父表中使用uselist参数来表示。

  • 一对多改写成一对一

    1
    class Parent(Model):
    2
        __tablename__ = 'parent'
    3
        id = Column(Integer, primary_key=True)
    4
        name = Column(String(64), nullable=False)
    5
        children = relationship('Child',uselist=False, back_populates='parent')
    6
    7
        def __repr__(self):
    8
            return "<Parent(name={})>".format(self.name)
    9
    10
    class Child(Model):
    11
        __tablename__ = 'child'
    12
        id = Column(Integer, primary_key=True)
    13
        email = Column(String(128), nullable=False)
    14
        parent_id = Column(Integer, ForeignKey('parent.id'))
    15
        parent = relationship('Parent', back_populates='children')
    16
    17
        def __repr__(self):
    18
            return "<Child(email={})>".format(self.email)

  • 多对一改写一对一

    1
    class Parent(Model):
    2
        __tablename__ = 'parent'
    3
        id = Column(Integer, primary_key=True)
    4
        name = Column(String(64), nullable=False)
    5
        children = relationship('Child', back_populates='parent')
    6
    7
        def __repr__(self):
    8
            return "<Parent(name={})>".format(self.name)
    9
    10
    class Child(Model):
    11
        __tablename__ = 'child'
    12
        id = Column(Integer, primary_key=True)
    13
        email = Column(String(128), nullable=False)
    14
        parent_id = Column(Integer, ForeignKey('parent.id'))
    15
        parent = relationship('Parent', back_populates='children', uselist=False)
    16
    17
        def __repr__(self):
    18
            return "<Child(email={})>".format(self.email)

  • 使用backref

    1
    from sqlalchemy.orm import backref
    2
    3
    class Parent(Model):
    4
        __tablename__ = 'parent'
    5
        id = Column(Integer, primary_key=True)
    6
        name = Column(String(64), nullable=False)
    7
        children = relationship('Child', backref=backref('parent', uselist=False))
    8
    9
        def __repr__(self):
    10
            return "<Parent(name={})>".format(self.name)
    11
    12
    13
    class Child(Model):
    14
        __tablename__ = 'child'
    15
        id = Column(Integer, primary_key=True)
    16
        email = Column(String(128), nullable=False)
    17
        parent_id = Column(Integer, ForeignKey('parent.id'))
    18
    19
        def __repr__(self):
    20
            return "<Child(email={})>".format(self.email)

多对多

在多对多关系中，A 表中的一行可以匹配 B 表中的多行，反之亦然。要创建这种关系，需要定义第三个表，称为结合表，它的主键由 A 表和 B 表的外部键组成。

多对多关系会在两个类之间增加一个关联的表。
这个关联的表在 relationship() 方法中通过 secondary 参数来表示。
通常的，这个表会通过 MetaData 对象来与声明基类关联。
所以这个 ForeignKey 指令会使用链接来定位到远程的表。

1
## 多对多关系的两个表之间的一个关联表
2
from sqlalchemy import Table, Text
3
association_table = Table('association', Model.metadata, 
4
                    Column('left_id', Integer, ForeignKey('left.id')),
5
                    Column('right_id', Integer, ForeignKey('right.id'))
6
                    )
7
class Parent(Model):
8
    __tablename__ = 'left'
9
    id = Column(Integer, primary_key=True)
10
    children = relationship('Child', secondary=association_table)
11
12
class Child(Model):
13
    __tablename__ 'right'
14
    id = Column(Integer, primary_key=True)

subquery load

1
from sqlalchemy.orm import subqueryload
2
3
# SQL: SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password FROM users WHERE users.name = 'jusene'
4
# SQL: SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id, anon_1.users_id AS anon_1_users_id FROM (SELECT users.id AS users_id FROM users WHERE users.name = 'jusene') AS anon_1 INNER JOIN addresses ON anon_1.users_id = addresses.user_id ORDER BY anon_1.users_id, addresses.id
5
jusene = session.query(User).options(subqueryload(User.addresses)).filter(User.name=='jusene').one()
6
jusene
7
<User(name=jusene, fullname=Jusene Ben)
8
jusene.addresses
9
[<Address(email_address=jusene@123.com)>,
10
 <Address(email_address=jusene@163.com)>]

在本例中，我们希望指明该用户。地址应尽快加载。加载一组对象及其相关集合的一个很好的选择是orm.subqueryload()选项，它会发出第二个SELECT语句，该语句会完全加载与刚刚加载的结果相关联的集合。

joined load

1
from sqlalchemy.orm import joinedload
2
3
# SQL: SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password, addresses_1.id AS addresses_1_id, addresses_1.email_address AS addresses_1_email_address, addresses_1.user_id AS addresses_1_user_id FROM users LEFT OUTER JOIN addresses AS addresses_1 ON users.id = addresses_1.user_id WHERE users.name = 'jusene' ORDER BY addresses_1.id
4
jusene = session.query(User).options(joinedload(User.addresses)).filter(User.name == 'jusene').one()
5
jusene
6
<User(name=jusene, fullname=Jusene Ben)
7
jusene.addresses
8
[<Address(email_address=jusene@123.com)>,
9
 <Address(email_address=jusene@163.com)>]

原文作者：Zhang Jusene

原文链接：http://jusene.github.io/2018/10/07/sqlalchemy-1/

发表日期：October 7th 2018, 1:43:33 pm

更新日期：November 30th 2019, 1:54:24 am

版权声明：本文采用知识共享署名-非商业性使用 4.0 国际许可协议进行许可

CATALOG
  1. 1. 建立关系
  2. 2. 连表查询
    1. 2.1. 使用别名
    2. 2.2. 使用子查询
    3. 2.3. 使用exists
  3. 3. 表关系
    1. 3.1. 一对多
    2. 3.2. 多对一
    3. 3.3. 一对一
    4. 3.4. 多对多
  4. 4. subquery load
  5. 5. joined load
Total : 190
2020
2019
2018
2017
python boot troubshooting docker drbd ssh web ELK kubernetes Kubernetes git ansible tools dns corosync consul Django etcd security ftp Go bash bigdata grub heartbeat Java kickstart kvm lvs mongodb mysql mesh nc nfs mogilefs openresty upgrade puppet rabbitmq saltstack rsync screen socat svn tcpdump fastdfs haproxy iscsi keepalived cache openssl vpn optimization prometheus redis RPM cncf supervisord kernel JAVA
python Linux troubshooting Docker docker storage nginx Nosql kubernetes Flask Kubernetes ansible Shell cluster Django Go bigdata boot Java hypervisor mysql mesh tools openresty puppet rabbitmq saltstack optimization cncf varnish tomcat JAVA