（四）SQLAlchemy相关对象和装载机策略_sqlalchemy.orm registry

一： 数据准备：

• 创建两个表： user和address

  from sqlalchemy.orm import registry, relationship
  from sqlalchemy import Column, Integer, ForeignKey, String
  from main import engine
  
  my_registry = registry()
  Base = my_registry.generate_base()
  
  
  class User(Base):
      __tablename__ = 'users'
      id = Column(Integer, primary_key=True)
      name = Column(String(30))
      fullname = Column(String)
      addresses = relationship("Address", back_populates="user")
  
  
  class Address(Base):
      __tablename__ = "address"
      id = Column(Integer, primary_key=True)
      email_address = Column(String, nullable=False)
      user_id = Column(Integer, ForeignKey("users.id"))
      user = relationship("User", back_populates="addresses")
  
  
  my_registry.metadata.create_all(engine)
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26

• 插入数据到数据库：

  from sqlalchemy import create_engine
  from sqlalchemy.orm import Session
  import settings
  from module import User, Address
  
  engine = create_engine(settings.DB_URI, echo=True)
  session = Session(engine)
  
  user1 = User(name="梁", fullname="梁山")
  user2 = User(name="王", fullname="王泽")
  user1.addresses.append(Address(email_address="liangshan@163.com", user_id="1"))
  user2.addresses.append(Address(email_address="wangze@163.com", user_id="2"))
  session.add(user1)
  session.add(user2)
  
  session.commit()
  session.close()
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17

二： 持久化与加载关系

2.1： relationship的作用：

• 通过relationship可以通过属性的方式，查询关联的对象信息。
• 注意： 在一对多中，属性是多的一方会返回列表，但是属性是一的一方返回的会是对象。

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
import settings
from module import User, Address

engine = create_engine(settings.DB_URI, echo=True)
session = Session(engine)
# 1：查询数据：
user = session.query(User).filter(User.name == "梁").first()
# 2: 打印user关联的地址：
print(user.addresses[0].email_address)
# liangshan@163.com

address = session.query(Address).filter(Address.email_address == "liangshan@163.com").first()
print(address.user.fullname)
# 梁山
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• 我们还可以这样操作： 先实例化一个地址对象，然后添加给某个User对象，这样就可以让某个人直接有了某个地址了。

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
import settings
from module import User, Address

engine = create_engine(settings.DB_URI, echo=True)
session = Session(engine)

address = Address(email_address="rsw@qiqi.com")
user = User(name="李", fullname="李黄德")
user.addresses.append(address)

print(user.addresses)
# [<module.Address object at 0x7f9f2ed37f98>]
print(address.user)
# <module.User object at 0x7f1798b76b38>
session.add(user)
session.commit()
session.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

• 解释会话的级联添加：
  • 例如上面的user对象，通过属性的方式给addresses追加了数据，那么当将user对象加入到会话的时候，同时也会将address对象加载到会话中，如果user追加了多个addresses，则也会按照顺序加入多个address对象。

2.2： 加载关系：

• 1： 当我们提交会话后，再次访问对象的属性的时候，会触发一次查询全部信息的操作，之后再次访问这个对象的任何信息都不会触发数据库查询操作。

• 2： 访问的属性如果是关联属性则会查询出关联对象的信息，如果不使用则不会查询。（懒加载机制）

• 3： 案例：

  from sqlalchemy import create_engine
  from sqlalchemy.orm import Session
  import settings
  from module import User, Address
  
  engine = create_engine(settings.DB_URI, echo=True)
  session = Session(engine)
  
  address = Address(email_address="rsw@qiqi.com")
  user = User(name="李", fullname="李黄德")
  user.addresses.append(address)
  
  print(user.addresses)
  # [<module.Address object at 0x7f9f2ed37f98>]
  print(address.user)
  # <module.User object at 0x7f1798b76b38>
  session.add(user)
  session.commit()
  
  # 关闭后再次使用属性：
  print(user.id)
  print(user.addresses)
  
  # 再次属性访问，我们发现不会向数据库查询数据，而是直接拿到了。
  print(user.id)
  print(user.id)
  
  session.close()
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28

• SQl的执行结果：

  • 结果显示： 提交之后，我们对user对象进行了三次属性的访问，但是只查询的一次。
  • 第二次使用user.address才触发了address的查询，而上次的user.name并没有触发关联查询。

  2021-08-24 09:50:05,780 INFO sqlalchemy.engine.Engine COMMIT
  2021-08-24 09:50:05,810 INFO sqlalchemy.engine.Engine BEGIN (implicit)
  2021-08-24 09:50:05,811 INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname 
  FROM users 
  WHERE users.id = %(pk_1)s
  2021-08-24 09:50:05,811 INFO sqlalchemy.engine.Engine [generated in 0.00018s] {'pk_1': 6}
  6
  2021-08-24 09:50:05,870 INFO sqlalchemy.engine.Engine SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id 
  FROM address 
  WHERE %(param_1)s = address.user_id
  2021-08-24 09:50:05,870 INFO sqlalchemy.engine.Engine [generated in 0.00019s] {'param_1': 6}
  [<module.Address object at 0x7f393a5c3fd0>]
  6
  6
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14

2.3： N+1问题：

• 根据上面的案例： 假设我们要查询10条user的信息，那么每个user又会查询一次address。查询10条user我们只需要1条查询语句，但是这个10个user对象每个又会产生一个查询语句，因此10 + 1=11条查询语句。

• 案例： 现在我数据库有三天user记录： 我查询所有的user，看看返回的信息。

• 懒加载模式是： 用到了我再加载，用不到我不加载。例如下面的案例： 如果不使用user.addresses，而是使用其他的属性。并不会触发那N次查询。

  from sqlalchemy import create_engine
  from sqlalchemy.orm import Session
  import settings
  from module import User
  
  engine = create_engine(settings.DB_URI, echo=True)
  session = Session(engine)
  # 演示N + 1问题：此时只会产生一个查询语句
  users = session.query(User).all()
  
  for user in users:
      print(user.addresses)
      # 发出三条SQL查询语句
  
  session.close()
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15

• N + 1：带来的问题：

  • 带来额外的查询语句，并且这些查询语句是隐式的，不容易被发现。

三： 装载机策略

3.1： selectinload装载机：

• 1：解决的问题: 对于想要得到一个不需要关联属性的对象，我们可以使用装载机来得到，这个对象不能访问关联属性，一旦访问就会报错。

  from sqlalchemy import create_engine, select
  from sqlalchemy.orm import Session, selectinload
  import settings
  from module import User
  
  engine = create_engine(settings.DB_URI, echo=True)
  session = Session(engine)
  
  
  users = session.execute(select(User).options(selectinload(User.addresses)))
  # 2021-08-24 10:54:03,917 INFO sqlalchemy.engine.Engine SELECT users.id, users.name, users.fullname
  # FROM users
  for user in users:
      print(user.addresses)
      # 报错
  
  session.close()
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17

• 2： 除了在查询上使用装载机，我们还可以在模型类上定义装载机。

  • 注意： 与查询不同的是，我们仍然可以访问address，只不过装载机进行行了优化。对于同一个session中，会将sql语句合并成一个sql语句。
  • 下面案例查看前后的执行结果。

  class User(Base):
      __tablename__ = 'users'
      id = Column(Integer, primary_key=True)
      name = Column(String(30))
      fullname = Column(String)
      addresses = relationship("Address", back_populates="user", lazy="selectin")
  1
  2
  3
  4
  5
  6

  engine = create_engine(settings.DB_URI, echo=True)
  session = Session(engine)
  
  users = session.query(User).all()
  for user in users:
      print(user.addresses[0].email_address)
  
  session.close()
  1
  2
  3
  4
  5
  6
  7
  8

  # 使用装载机之前：
  2021-08-24 11:01:57,934 INFO sqlalchemy.engine.Engine BEGIN (implicit)
  2021-08-24 11:01:57,935 INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname 
  FROM users
  2021-08-24 11:01:57,936 INFO sqlalchemy.engine.Engine [generated in 0.00015s] {}
  2021-08-24 11:01:57,995 INFO sqlalchemy.engine.Engine SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id 
  FROM address 
  WHERE %(param_1)s = address.user_id
  2021-08-24 11:01:57,995 INFO sqlalchemy.engine.Engine [generated in 0.00017s] {'param_1': 1}
  liangshan@163.com
  2021-08-24 11:01:58,026 INFO sqlalchemy.engine.Engine SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id 
  FROM address 
  WHERE %(param_1)s = address.user_id
  2021-08-24 11:01:58,026 INFO sqlalchemy.engine.Engine [cached since 0.03053s ago] {'param_1': 2}
  wangze@163.com
  2021-08-24 11:01:58,055 INFO sqlalchemy.engine.Engine SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id 
  FROM address 
  WHERE %(param_1)s = address.user_id
  2021-08-24 11:01:58,055 INFO sqlalchemy.engine.Engine [cached since 0.06011s ago] {'param_1': 3}
  rsw@qiqi.com
  2021-08-24 11:01:58,085 INFO sqlalchemy.engine.Engine ROLLBACK
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21

  # 使用装载机之后： 
  2021-08-24 11:06:04,317 INFO sqlalchemy.engine.Engine BEGIN (implicit)
  2021-08-24 11:06:04,318 INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname 
  FROM users
  2021-08-24 11:06:04,318 INFO sqlalchemy.engine.Engine [generated in 0.00015s] {}
  2021-08-24 11:06:04,378 INFO sqlalchemy.engine.Engine SELECT address.user_id AS address_user_id, address.id AS address_id, address.email_address AS address_email_address 
  FROM address 
  WHERE address.user_id IN (%(primary_keys_1)s, %(primary_keys_2)s, %(primary_keys_3)s)
  2021-08-24 11:06:04,378 INFO sqlalchemy.engine.Engine [generated in 0.00029s] {'primary_keys_1': 1, 'primary_keys_2': 2, 'primary_keys_3': 3}
  liangshan@163.com
  wangze@163.com
  rsw@qiqi.com
  2021-08-24 11:06:04,409 INFO sqlalchemy.engine.Engine ROLLBACK
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13

3.2： joinedload装载机

• 善于处理多对一对象： 直接将一的哪一方的属性拿过来使用。
• joinload如果不指定innerjoin=True，默认是使用左外连接。

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, joinedload
import settings
from module import User, Address

engine = create_engine(settings.DB_URI, echo=True)
session = Session(engine)

addresses = session.execute(select(Address).options(joinedload(Address.user, innerjoin=True)))
# 2021-08-24 11:32:46,009 INFO sqlalchemy.engine.Engine SELECT address.id, address.email_address, address.user_id, users_1.id AS id_1, users_1.name, users_1.fullname
# FROM address JOIN users AS users_1 ON users_1.id = address.user_id
for address in addresses:
    print(address[0].user_id, address[0].user.name, address[0].user.fullname)
    # 1 梁 梁山
    # 2 王 王泽
    # 3 李 李黄德
    
session.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

2021-08-24 11:34:55,455 INFO sqlalchemy.engine.Engine
        SELECT address.id, address.email_address, address.user_id, users_1.id AS id_1, users_1.name, users_1.fullname 
FROM address JOIN users AS users_1 ON users_1.id = address.user_id
1
2
3

• 显示连接：一般我们写代码的时候为了显示哪两个表连接，我们会显示的进行连接，但是此时存在了一个问题会导致两次进行join连接。

engine = create_engine(settings.DB_URI, echo=True)
session = Session(engine)

addresses = session.execute(select(Address).join(Address.user).options(joinedload(Address.user, innerjoin=True)))
# 2021-08-24 11:54:09,372 INFO sqlalchemy.engine.Engine SELECT address.id, address.email_address, address.user_id, users_1.id AS id_1, users_1.name, users_1.fullname 
# FROM address JOIN users ON users.id = address.user_id JOIN users AS users_1 ON users_1.id = address.user_id
session.close()
1
2
3
4
5
6
7

• 为了解决这个问题，我们可以用 contains_eager(Address.user)， 去掉一个连接。

engine = create_engine(settings.DB_URI, echo=True)
session = Session(engine)

addresses = session.execute(select(Address).join(Address.user).options(contains_eager(Address.user)))
# 2021-08-24 11:58:12,897 INFO sqlalchemy.engine.Engine SELECT users.id, users.name, users.fullname, address.id AS id_1, address.email_address, address.user_id 
# FROM address JOIN users ON users.id = address.user_id
session.close()
1
2
3
4
5
6
7

3.3： raiseload装载机：

• 被这个装载机装载后，将阻止所有的SQL加载操作：

class User(Base):
    __tablename__ = 'user_account'

    # ... Column mappings

    addresses = relationship("Address", back_populates="user", lazy="raise_on_sql")


class Address(Base):
    __tablename__ = 'address'

    # ... Column mappings

    user = relationship("User", back_populates="addresses", lazy="raise_on_sql")
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 使用关系属性，不会触发底层的查询语句，而是被阻断。
u1 = s.execute(select(User)).scalars().first()
u1.addresses
sqlalchemy.exc.InvalidRequestError: 'User.addresses' is not available due to lazy='raise_on_sql'
1
2
3
4