Класс scoped_session

Класс scoped_session создает потокобезопасную обертку над сессией, которая автоматически привязывает сессию к текущему потоку выполнения. Первым параметром класс принимает фабрику сессий sessionmaker. Это позволяет использовать одну и ту же переменную сессии в разных потоках без конфликтов.

Синтаксис

scoped_session(session_factory, [scopefunc])

Пример

Давайте создадим scoped-сессию на основе фабрики sessionmaker:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

engine = create_engine('sqlite:///:memory:')
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

print(Session)

Результат выполнения кода:

"<sqlalchemy.orm.scoping.scoped_session object at ...>"

Пример

Давайте добавим запись в таблицу articles через scoped-сессию:

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(primary_key=True)
	title: Mapped[str]
	text: Mapped[str]
	status: Mapped[str]
	num: Mapped[int]

Теперь создадим таблицы и сохраним статью:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from models import Base, Article

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

article = Article(
	title='article',
	text='text of article',
	status='published',
	num=1
)

Session.add(article)
Session.commit()

res = Session.query(Article).all()
print(res[0].title)

Результат выполнения кода:

"article"

Пример

Давайте проверим, что scoped-сессия возвращает один и тот же объект в рамках одного потока:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

engine = create_engine('sqlite:///:memory:')
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

session_one = Session()
session_two = Session()

print(session_one is session_two)

Результат выполнения кода:

True

Смотрите также

класс scoped_session,
который создает потокобезопасную сессию
метод remove,
который удаляет текущую сессию из реестра
метод configure,
который перенастраивает фабрику сессий
метод query,
который создает запрос через scoped-сессию