在这篇文章中，我们查看了SessionMaker，Scoped_session及其关联的查询方法的一些基础知识。
请注意，这篇文章不是教程。我试图回答我自己的一些问题。我只是写下我的答案。
本文中使用的源数据库是Oracle Corporation发布的 MySQL测试数据。可从https://github.com/datacharmer/test_db。
下载让我们从会话课开始。 sqlalchemy官方文档Using the Session。
¶我们可以使用 sessionmaker 的实例运行全文查询：
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text
SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://behai:pcb.2176310315865259@localhost/employees"
engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=False, future=True)
session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)
session = session_factory()
sql = "select * from employees where last_name like '%treh%' limit 0, 10"
dataset = session.execute(text(sql))
for r in dataset:
print(r)
dataset.close()
session.close()
对于 future = true ，请参阅SQLAlchemy 2.0 Future (Core)
- 接下来是 scoped_session 。基本上，这是我们在Web应用程序中使用的会话：每个范围的会话都是在Web请求的上下文中的本地。请参阅Contextual/Thread-local Sessions。
出于讨论的目的，我们将不会执行任何Web应用程序。简单的命令行Python脚本足以说明目的：
from threading import get_ident
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy import text
SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://behai:pcb.2176310315865259@localhost/employees"
engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=False, future=True)
session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)
database_sesssion = scoped_session(session_factory, scopefunc=get_ident)
session = database_sesssion(future=True)
sql = "select * from employees where last_name like '%treh%' limit 0, 10"
dataset = session.execute(text(sql))
for r in dataset:
print(r)
dataset.close()
session.close()
此脚本与上一个脚本相同，除非增加一个，
和一个修改：
加法是第12行：
database_sesssion = scoped_session(session_factory, scopefunc=get_ident)
对于 scopefunc = get_ident ，请参阅sqlalchemy.orm.scoping.scoped_session，此stackoverflow post flask app_ctx_stack.ident_func_ error due to ident_func deprecated in werkzeug 2.1应该更清楚。
修改为第14行：
session = database_sesssion(future=True)
在第二个脚本中，我们不是直接从 sessionmaker 获得会话，而是从 scoped_session 间接获得一个会话。我们正在运行与上一个脚本相同的查询，因此最终输出也相同。
请注意：
从这里开始，脚本都有第1-12行（一到十二）相同，我将仅列出当前讨论的相关新代码。
根据Contextual/Thread-local Sessions的根据：session = database_sesssion(future=True)
如果我们重复调用 database_sesssion（），我们将获得相同的会话：
...
session = database_sesssion(future=True)
session1 = database_sesssion()
print(f"1. database_sesssion: {id(database_sesssion)}")
print(f"1. session: {id(session)}")
print(f"1. session1: {id(session1)}")
print(f"1. session is session1: {session is session1}")
1. database_sesssion: 1724058633408
1. session: 1724061992896
1. session1: 1724061992896
1. session is session1: True
我们可以调用 database_sesssion（future = true）仅一次，必须在没有参数的情况下进行后续呼叫，否则将导致以下例外：
(venv) F:\my_project>venv\Scripts\python.exe src\my_project\my_python_script.py
Traceback (most recent call last):
File "F:\my_project\src\my_project\my_python_script.py", line 15, in <module>
session1 = database_sesssion(future=True)
File "F:\my_project\venv\lib\site-packages\sqlalchemy\orm\scoping.py", line 39, in __call__
raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Scoped session is already present; no new arguments may be specified.
根据method sqlalchemy.orm.scoping.scoped_session.remove()：
如果存在，则处置当前会话。
这将在当前会话中首先调用session.close（）方法，该方法释放了仍在持有的任何现有的交易/连接资源；交易专门回滚。然后会丢弃会话。在同一范围内的下一个用法后，scoped_session将产生一个新的会话对象。
让我们看看这意味着什么，以下脚本产生与第一个脚本相同的结果：
...
session = database_sesssion(future=True)
session.close()
database_sesssion.remove()
session = database_sesssion(future=True)
sql = "select * from employees where last_name like '%treh%' limit 0, 10"
dataset = session.execute(text(sql))
for r in dataset:
print(r)
dataset.close()
session.close()
呼叫 database_sesssion.remove（），随后的参数调用 session = database_sesssion（future = true）实际上有效。 p>
这在上述语句的上下文中是有道理的：内部注册表现在为空，没有活动的范围会话，因此我们可以使用我们认为合适的任何配置创建一个新的。 这表明 database_sesssion 本身在调用 database_sesssion.remove（）之后仍然是同一对象。让我们用以下脚本进行测试： 我们可以看到它是在输出中： - 我们将引入一个模型，该模型基本上是数据库表的SQLalchemy类表示。根据文档，我们将从function sqlalchemy.orm.declarative_base(...)降下模型，然后我们可以将method sqlalchemy.orm.scoping.scoped_session.query_property(query_cls=None)与此模型一起使用。我们将为员工做一个模型表。 ❺⓵尽管有官方的文档示例，我还是想尝试使用 database_sesssion.query_property（）直接，这是我的第一个尝试（我列出了一个完整的新脚本）： 我确实期望它能起作用，但没有： 从现在开始，脚本将具有与❺⓵相同的第1-22行（一到两十二行），我将仅列出当前讨论的相关添加和更改。 ❺⓶在method sqlalchemy.orm.scoping.scoped_session.query_property(query_cls=None)下列出的示例摘要是： 因此，我修改❺⓵如下： 确实可以按预期工作： 我不知道为什么必须这样。但是官方文件指出必须是，我的实验证实了这一点。我不会研究Sqlalchemy代码以了解原因！我觉得这也将是徒劳的，也... 我们之前已经看到 scoped_session（...）（[...]）和 sessionmaker（...）（）（）都导致class sqlalchemy.orm.Session(...) ，这堂课有一个method sqlalchemy.orm.Session.query(entities, **kwargs)。此方法可用于执行** scoped_session（...）。query_property（）*可以做一些差异。首先是该模型可以直接从 dem> dectarative_base（）下降。其次，语法略有不同。 ❺⓷使用 scoped_session（...）（[...]）'s query（）方法： ❺⓸，同样适用于 sessionmaker（...）（）： 让我们重申，我们将 scoped_session 与Web应用程序-Contextual/Thread-local Sessions一起使用。我们包括 SessionMaker ，以摇动比较和完整性。返回到 scoped_session的query（）方法和查询属性： ❺⓶： result = employ.query.filter（employ.emp_no == 16621）.first（） ❺⓷：结果= session.query（雇员）.filter（employ.emp_no == 10545）.first（） 我选择使用❺⓶方法，尽管必须有 this basemodel 可以是所有数据库表的父类，因此从长远来看，它可以自行付款。 €â€ 正如我所说的那样，这篇文章不是教程。我发现Sqlalchemy是一个艰难的行程...我着手回答自己的问题。我希望这篇文章的内容对学习Sqlalchemy的其他人有用。我确实希望我在这篇文章中没有犯任何错误。感谢您的阅读和保持安全。
...
session = database_sesssion(future=True)
session1 = database_sesssion()
print(f"1. database_sesssion: {id(database_sesssion)}")
print(f"1. session: {id(session)}")
print(f"1. session1: {id(session1)}")
print(f"1. session is session1: {session is session1}")
session1.close()
session.close()
database_sesssion.remove()
session = database_sesssion(future=True)
session1 = database_sesssion()
print(f"2. database_sesssion: {id(database_sesssion)}")
print(f"2. session: {id(session)}")
print(f"2. session1: {id(session1)}")
1. database_sesssion: 2102627796160
1. session: 2102631155648
1. session1: 2102631155648
1. session is session1: True
2. database_sesssion: 2102627796160
2. session: 2102631155792
2. session1: 2102631155792
from threading import get_ident
from sqlalchemy import (
create_engine,
Column,
Integer,
Date,
String,
)
from sqlalchemy.orm import sessionmaker, scoped_session
"""
Any one of these import of declarative_base will work.
"""
# from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import declarative_base
SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://behai:pcb.2176310315865259@localhost/employees"
engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=False, future=True)
session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)
database_sesssion = scoped_session(session_factory, scopefunc=get_ident)
class Employees(declarative_base()):
__tablename__ = 'employees'
emp_no = Column(Integer, primary_key=True)
birth_date = Column(Date, nullable=False)
first_name = Column(String(14), nullable=False)
last_name = Column(String(16), nullable=False)
gender = Column(String(1), nullable=False )
hire_date = Column(Date, nullable=False )
query = database_sesssion.query_property()
result = query.filter(Employees.emp_no==16621).first()
(venv) F:\my_project>venv\Scripts\python.exe src\my_project\my_python_script.py
Traceback (most recent call last):
File "F:\my_project\src\my_project\my_python_script.py", line 35, in <module>
result = query.filter(Employees.emp_no==16621).first()
AttributeError: 'query' object has no attribute 'filter'
请注意：
<pre>
Session = scoped_session(sessionmaker())
class MyClass(object):
query = Session.query_property()
# after mappers are defined
result = MyClass.query.filter(MyClass.name=='foo').all()
...
class BaseModel(object):
query = database_sesssion.query_property()
Base = declarative_base(cls=BaseModel)
class Employees(Base):
__tablename__ = 'employees'
...
result = Employees.query.filter(Employees.emp_no==16621).first()
print(result.__dict__)
(venv) F:\my_project>venv\Scripts\python.exe src\my_project\my_python_script.py
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x0000026A680079A0>, 'last_name': 'Strehl', 'emp_no': 16621, 'hire_date': datetime.date(1992, 6, 11), 'first_name': 'Parviz', 'gender': 'M', 'birth_date': datetime.date(1962, 5, 30)}
...
class Employees(declarative_base()):
__tablename__ = 'employees'
...
session = database_sesssion(future=True)
result = session.query(Employees).filter(Employees.emp_no==10545).first()
print( result.__dict__ )
...
class Employees(declarative_base()):
__tablename__ = 'employees'
...
session = session_factory()
result = session.query(Employees).filter(Employees.emp_no==11000).first()
print( result.__dict__ )
额外的基础课。即：
class BaseModel(object):
query = database_sesssion.query_property()