/
romaaaka
/
queuebot
Обзор
Документация
Войти
/
romaaaka
/
queuebot
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/api/dependencies.py
26 строк
828 B
Roman
more safety locks
09 май 2025, 15:35
09 май 2025, 15:35
9590f5c
Код
Авторство
О чём код?
from typing import Annotated from fastapi import Depends from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from api.application import alchemy from containers import ServicesContainer DatabaseSession = Annotated[AsyncSession, Depends(alchemy.provide_session())] async def get_services( session: DatabaseSession, ) -> ServicesContainer: # FastAPI вызовет alchemy.provide_session ровно один раз за запрос # Устанавливаем таймаут захвата блокировок в 5 секунд await session.execute(text("SET lock_timeout = '10s'")) try: yield ServicesContainer(session) await session.commit() except Exception: await session.rollback() raise finally: await session.close()