python常用库之数据库orm框架之SQLAlchemy

python常用库之数据库orm框架之SQLAlchemy

一、什么是SQLAlchemy

官网：https://www.sqlalchemy.org/
SQLAlchemy是一个基于Python实现的SQL工具包和ORM框架,提供了高层抽象来管理数据库交互。
SQLAlchemy功能强大,可以省去很多手动管理数据库连接、资源、事件等重复工作,让开辟者更加高效地使用数据库。许多大型Python项目都选择使用SQLAlchemy作为ORM框架。
SQLAlchemy 使用场景

SQLAlchemy是一个强大的Python ORM框架,紧张应用于以了局景:

• 数据库访问和操作:SQLAlchemy提供了高层抽象来操作数据库,可以制止写原生SQL语句。支持多种数据库后端。
  
• ORM映射:建立Python类与数据库表的映射关系,简化数据模子的操作,支持声明式操作。
  
• 复杂查询: SQLAlchemy提供丰富的查询方式,如过滤、分组、联结等,可以构建复杂查询。
  
• 异步查询:基于Greenlet等实现异步查询,提高查询效率。
  
• 事件控制: 通过Session管理数据库会话和事件。
  
• 工具集成:如数据迁移工具Alembic,可以实现Schema版本控制和迁移。
  
• 大数据集查询:基于Pagination实现数据分页,制止大量数据查询内存溢出。
  
• 多数据库支持:支持Postgres、MySQL、Oracle等主流数据库。
  
• Web框架集成:框架如Flask可以集成SQLAlchemy,便于Web应用开辟。
  

二、SQLAlchemy使用

SQLAlchemy根据模子查询

• 引入和初始化
  导入sqlalchemy模块,然后使用create_engine创建引擎,sessionmaker创建会话类并生成会话对象。
  

1. from sqlalchemy import create_engine
2. from sqlalchemy.orm import scoped_session, sessionmaker
4. engine = create_engine(SQLALCHEMY_DATABASE_URI)
5. db_session = scoped_session(sessionmaker(autoflush=False,bind=engine))

复制代码

• 使用sql 或者 数据模子
  

这里先举例模子方式：
在SQLAlchemy中,declarative_base()用来创建一个基类,这个基类会为所有继续它的子类提供declarative的ORM功能。
TblObject = declarative_base() 的作用是:
3. 创建了一个名为TblObject的基类
4. 这个TblObject具有declarative的功能
5. 后续定义的模子类可以继续这个TblObject基类来使用declarative

1. from sqlalchemy.orm import declarative_base
2. from sqlalchemy import Column, Text, Integer,String, VARCHAR, TIMESTAMP, BOOLEAN, Float, text
4. TblObject = declarative_base()
6. class User(TblObject):
7.         __tablename__ = 'table_user'
8.         __table_args__ = ({"schema": "public"})
9.         id = Column(Integer, primary_key=True)
10.         name = Column(String)
12. query = db_session.query(User)
13. user = query.first()
14. print(user.name)

复制代码

SQLAlchemy SQL 格式化的方式

• 将sql语句本身不包含参数格式化,直接作为文本构建完备的sql:
  要想实现变量替换,需要使用f-string的格式
  

1. myschema = 'xxxx'
2. sql = text(f'SELECT id FROM {myschema }.table...')

复制代码

注意看字符串前面有个f。

• 使用参数绑定的方式,而不是%格式化:
  

1. sql = text('SELECT id FROM :schema.table...')
2. params = {'schema': 'my_schema'}
3. result = session.execute(sql, params)

复制代码

总结:

• sql文本内不要再使用% formatting
  
• 使用f-string或参数绑定格式化
  
• 创建text对象时直接构建完备sql文本
  

   参数绑定也可以提高安全性,制止SQL注入\
  db_session.query和 db_session.execute区别

在SQLAlchemy中,db_session.query()和db_session.execute()紧张有以下几点区别:

• 返回值差别:
  
  
  
  • db_session.query() 返回一个Query对象,可以用于构建查询并最终获取结果。
    
  • db_session.execute() 直接实行语句并返回结果。
    
  
  
• 查询方式差别:
  
  
  
  • db_session.query() 通过ORM构建查询。
    
  • db_session.execute() 通过原生SQL语句查询。
    
  
  
• 查询灵活性差别:
  
  
  
  • db_session.query() 可以构建非常灵活的查询,包含各种过滤、JOIN等。
    
  • db_session.execute() 只能实行简朴的SQL语句查询。
    
  
  
• 返回结果差别:
  
  
  
  • db_session.query() 返回的是ORM对象或者自定义类的实例。
    
  • db_session.execute() 直接返回行数据构成的列表。
    
  
  
• 性能差别:
  
  
  
  • 对简朴查询,db_session.execute()每每更快。
    
  • 对复杂查询,db_session.query()可以通过ORM特性举行优化。
    所以,简朴来说:
    
  
  

• db_session.query() 是面向对象的查询方式,更灵活,但复杂查询可能有性能问题。
  
• db_session.execute() 是实行原生SQL的直接查询,性能好但不够灵活。
  

   需要特殊注意的点：使用 db_session.query() 后获取的 ORM 对象实例,在访问其属性之前,需要确保与该查询关联的 db_session 没有关闭或失效。
  否则报错：

1. raise orm_exc.DetachedInstanceError(
2. sqlalchemy.orm.exc.DetachedInstanceError: Instance <User at 0x7f2973d6fbe0> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)

复制代码

通常的使用模式是:

1. db_session = Session()
3. # 创建查询
4. query = db_session.query(User)
6. # 获取用户对象
7. user = query.first()
9. # 访问属性
10. print(user.name)
12. # 关闭会话
13. db_session.close()

复制代码

在关闭 db_session 之前,需要先完成对其查询结果的处理和访问。
如果在获取到 user 对象后立即关闭会话,然后访问 user.name,会发生 detached实例错误。
因为在会话关闭后,user 对象脱离了会话,属性无法加载。
   使用 db_session.execute() 实行原生SQL查询general不会存在会话过早关闭导致属性访问错误的问题。
db_session.execute() 返回的是数据库行记录构成的结果集,不涉及 ORM 对象映射。
  实测demo 总结：让我们注意一下SQLAlchemy 的 lazy loading 特性

测试结果：经过测试，query = db_session.query(User).all() 这样db_session.close() 后，使用query[0].name会报错，但是query = db_session.query(User)，close之后 query的数据还能用。
db_session.query(User) 只是构造了一个查询,并没有立即获取结果。
而 db_session.query(User).all() 实行了查询并获取告终果。
在会话关闭后,这两种环境的举动差别:

• 对于仅构造查询的 db_session.query(User),由于没有结果产生,所以不会有 detached 实例的问题。我们可以在会话关闭后,使用这个查询加上过滤条件等再次实行,产生新结果。
  
• 但对于立即实行了的 db_session.query(User).all(),其结果中的实例与关闭后的会话失去关联,成为 detached 实例,会导致访问属性失败。
  

这与 SQLAlchemy 的 lazy loading 特性有关 - 查询只有在需要时才实行和加载实例。

• 构造查询只生成了一个实行计划,不涉及详细实例数据,所以会话关闭后计划还可以再实行。
  
• 但实行查询并获取实例,实际上已经加载了详细的数据,所以会依赖会话提供的数据状态。
  

scoped session

使用 db_session.query() 时,为了制止访问 detached 实例,我们希望耽误关闭会话 db_session。但这样就无法实时提交事件,可能会导致锁表问题。
办理方法：

• 在查询之后、访问实例属性之前,先显式提交事件:
  

1. db_session.query(...)
3. db_session.commit() # 提交事务
5. obj.some_attr # 访问属性

复制代码

这可以先释放锁,同时实例也绑定到会话中。

• 使用scoped session,可以制止手动 close db_session。
  

1. from sqlalchemy.orm import sessionmaker, scoped_session
3. session_factory = sessionmaker(bind=engine)
4. Session = scoped_session(session_factory)
6. def get_user():
7.     session = Session()
8.     return session.query(User).first()

复制代码

这里的Session是一个scoped session类。其特点是:

• 每个线程或哀求都会主动创建一个新的session实例,制止同一个session跨线程/哀求使用。
  
• 使用完session后不需要关闭它,scoped session会在当前上下文退出后主动关闭、invalidate该session。
  
• 所以我们只需要使用Session类创建session,不需要close。
  
• 差别的框架可以集成scoped session到本身的上下文。比如Flask集成后,每个哀求都会主动开启一个session,哀求结束后主动关闭。
  
• 这样就可以不需要我们手动管理session的生命周期。
  

所以综上,scoped session通过封装context管理,让session的生命周期与当前上下文(线程、哀求等)绑定,主动开启和关闭,省去了手动管理的问题。
   总结：
方法1: 我们只调commmit，这里不调colse()，需要其他地方，只有在确定不用的环境下才调close。
方法2：我们使用scoped_session，scoped session 的生命周期由应用管理,不需要我们手动关闭。
    使用scoped_session,这个 session 实例绑定到线程/哀求,在其线程结束时主动关闭.
  关于scoped_session主动提交事件

scoped_session 的工作原理是:

• 为每个线程或哀求创建一个新的 session 实例
  
• 这个 session 实例绑定到线程/哀求,在其结束时主动关闭
  
• 开辟者只需要使用 Session 类即可,不需要手动关闭
  也就是说,每个线程/哀求都会有一个独立的 session 实例,这个实例不会主动提交。
  

如果我们需要主动提交,需要在使用 session 的时候设置:
在会话层面设置主动提交:

1. from sqlalchemy.orm import sessionmaker
3. Session = sessionmaker(bind=engine)
4. session = Session()
6. session.autocommit = True

复制代码

通过设置session.autocommit启用主动提交。这会针对当前线程的 session 实例开启主动提交。
另外,scoped_session 也不发起使用主动提交模式,因为会有并发问题。推荐的方式还是手动提交控制事件。
总结,对于 scoped_session:

• 在创建时设置autocommit 不会收效
  
• 应在使用时针对 session 实例设置autocommit
  
• 推荐的方式仍是手动提交控制事件
  

参数autoflush=False 的工作机制

1. db_session = scoped_session(sessionmaker(autoflush=False, bind=engine))
2. db_session = scoped_session(sessionmaker(bind=engine))

复制代码

如上，有什么区别？
关闭主动 flush(autoflush=False)的紧张目的是可以更精细地控制 flush 的机遇,原因如下:

• 主动 flush 可能导致不须要的数据库写入操作,比如在一个事件中有多次 CRUD 操作时,希望只在事件提交时一次性写入,而不是每次操作都触发 flush。
  
• 在一些场景下需要确保 flush 只在事件提交时发生,如两个关联对象的变更,希望它们的变更作为一个事件实行。
  
• 需要根据业务逻辑准确控制 flush 机遇,而不是通过隐式的主动 flush。
  

SQLAlchemy中的session.commit()内部会主动调用session.flush()来保证所有pending的变更都被持久化。
即使设置autoflush=False,commit操作也会强制实行flush。只是在commit之外的其他环境下,需要手动调用flush。
在autoflush=False的环境下:

• 正常的增编削操作不会主动flush
  
• 调用commit会触发强制flush以持久化变更
  这是SQLAlchemy的一种掩护机制,来确保在事件提交时不会丢失还未flush的变更。
  

autoflush=False 需要手动flush,但不影响commit的主动flush举动。
三、关于SQLAlchemy主动提交事件

配景：
工作发现使用，出现很多 sqlalchemy出现很多未提交事件 环境。
demo：

1. from sqlalchemy import create_engine
2. from sqlalchemy.orm import scoped_session, sessionmaker
3. from sqlalchemy import text
6. username="aaa"
7. password="xxx"
8. host="127.0.0.1"
9. port="5432"
10. db="xx"
11. sqlalchemy_uri = f"postgresql://{username}:{password}@{host}:{port}/{db}"
13. engine = create_engine(sqlalchemy_uri)
15. #engine = create_engine(sqlalchemy_uri, echo=False, client_encoding='utf8', pool_size=100, pool_recycle=3600)
16. db_session = scoped_session(sessionmaker(bind=engine))
17.                                                                                                                                                   
18. sql = text('SELECT * FROM auth_user WHERE id = :userid;')
19. userid=1
21. #  execute() 方法仅仅是执行查询,不会自动提交事务。
22. objs = db_session.execute(sql, {"userid": userid}).fetchall()
23. print(objs)

复制代码

**每个session默认都是在一个事件中,不会主动提交。**因此当你调用db_session.execute(sql)时,会默认复用当前session的事件来实行,而这个session已经在一个未提交的事件中了。
这与直接调用begin()开始一个新事件其实结果是一样的。
PostgreSQL 和 MySQL 主动提交区别

PostgreSQL 的默认设置是主动提交关闭的,每次事件需要显式地通过 COMMIT 提交。
PostgreSQL的事件autocommit是transaction级的。每个transaction需要显式提交或回滚。
在PostgreSQL中,autocommit是作用于每个事件(transaction)的,紧张体现在以下两个方面:

• 新建立的连接默认都黑白主动提交状态(autocommit=off)。这个连接下的所有事件操作默认都黑白主动提交的。
  
• 开始一个新的事件时,这个事件会继续连接当前的autocommit状态。如果连接黑白主动提交的,那么这个事件也黑白主动提交的。
  

举个例子:
– 新建立连接,默认autocommit=off

1. postgres=# begin; -- 开始一个新事务,继承连接的autocommit=off
3. postgres=# insert into table_a values (1); -- 非自动提交
5. postgres=# commit; -- 需要显式提交事务

复制代码

  新连接默认的黑白主动提交状态。当开始一个新的事件时,这个事件继续了连接的非主动提交属性。所以我们必须通过提交事件来持久化改变。
  

• MySQL 的默认设置是主动提交开启的,每条语句会主动提交事件。
  MySQL的事件autocommit标志是session级的。也就是说,一个会话内全部操作默认都是主动提交的。
  

您的问题提到了MySQL中的主动提交模式,这和PostgreSQL中的事件处理模式有些差别。
MySQL中的主动提交(autocommit)是作用于整个数据库连接会话的,紧张体现在:

• 新建立的连接默认是主动提交模式(autocommit=on)。
  
• 连接的主动提交模式作用于该连接下实行的所有事件。不管实行多少个事件,都使用该连接的主动提交模式。
  
• 设置主动提交模式会改变连接的默认提交举动。
  

举个例子:
– 新连接默认autocommit=ON

1. mysql> start transaction; -- 开始事务,但继承连接的自动提交模式
3. mysql> insert into table_a values (1); -- 自动提交
5. mysql> commit; -- 提交无实际效果,因为已自动提交

复制代码

在这个例子中,由于连接的默认autocommit=on,所以不管实行多少个事件,每条SQL语句都会隐式提交。
SQLAlchemy 的计划是为了最大程度兼容差别数据库的举动,所以从 1.4 版本开始采用关闭主动提交作为默认值,这更符合 PostgreSQL 和一些其他数据库的举动。
总结：PostgreSQL的autocommit属性是针对每个事件的,而不是整个会话。这就要求我们必须显式地提交或回滚事件来结束一个事件。
查看PostgreSQL当前连接是否开启主动提交

查询 pg_settings 系统表:

1. SELECT setting FROM pg_settings WHERE name = 'default_transaction_isolation';

复制代码

返回 ‘read committed’ 表示未开启。
是否可以直接查询PostgreSQL的autocommit状态,而不通过 default_transaction_isolation 这种间接的方式。
直接查询autocommit的参数在PostgreSQL中是不支持的。因为autocommit实际上不是一个配置参数,而是个概念,是由default_transaction_isolation参数决定的。
没有办法直接用SHOW或者SELECT方式获取autocommit的状态。
之所以我们常用 default_transaction_isolation 来判断,是因为这两个设置在PostgreSQL内部是耦合的:

• read committed 表示关闭了autocommit
  
• read uncommitted 表示开启了autocommit
  所以default_transaction_isolation等于read committed的时候,就可以确定autocommit是关闭的。
  

四、工作遇到的问题

1. pg报错：unexpected EOF on client connection

在 PostgreSQL 中,“unexpected EOF on client connection” 这个错误通常表示客户端应用程序异常停止了与数据库的连接,导致有一个未完成的打开事件状态。
这种错误的常见原因包括:

• 应用程序进程崩溃或异常退出,没有正常关闭数据库连接。
  
• 网络连接异常停止。比如网络闪断,客户端机器宕机等。
  
• 客户端没有正确处理查询超时或者服务器重启的环境。
  
• 由于编程错误或者资源问题,客户端忘记提交/回滚一个长时间运行的事件。
  
• 数据库连接泄露,占用了所有可用连接。
  

当出现这个错误时,该未提交的事件会一直持有锁占用连接,壅闭其他事件,因此需要举行以下处理:

• 检查网络链接环境,确保客户端和数据库服务器网络畅通。
  
• 检查客户端应用程序代码,确保正确处理超时、事件提交等环境。
  
• 在数据库端实行以下下令,手动终止该占用连接的事件:
  

1. SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE state = 'idle in transaction';

复制代码

你实行这个语句后,处在 idle in transaction 状态的后端进程将被强制终止,相关的事件也会被回滚。
注：pg_terminate_backend() 函数的返回值 ‘t’ 表示终止后端乐成。
终止后端进程后,还需要几个后续步骤:

• 再次实行同样的查询,检查是否还有处在 idle in transaction 状态的进程:
  

1. SELECT * FROM pg_stat_activity WHERE state = 'idle in transaction';

复制代码

• 分析日志,找到未提交事件的根本原因,比如编程错误、资源问题等,从代码级别办理问题。
  
• 合理配置数据库连接池,制止连接泄露。
  
• 调解数据库配置,适当增大max_connections数值。
  

2. 报错：sqlalchemy.exc.ArgumentError: autocommit=True is no longer supported

早期版本的 SQLAlchemy 支持在 Engine 创建时设置 autocommit=True,将该引擎设置为主动提交事件模式。但是从版本 1.4 开始,这个参数就不再支持了。
导致这个错误的典型代码如:

1. engine = create_engine(URL, autocommit=True)

复制代码

要修复这个错误,需要移除 autocommit 参数,改为手动管理事件:

• 去掉 autocommit=True
  

1. engine = create_engine(URL)

复制代码

• 在代码中手动提交事件:
  

1. with engine.begin() as conn:
2.     conn.execute(...)
3.     conn.commit()

复制代码

• 或者开启主动提交模式:
  

1. connection = engine.connect()
2. connection.autocommit = True

复制代码

总之,Engine 对象不再支持 autocommit 参数。需要通过上述方式自行控制事件提交。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。