Эта статья рассчитана в большинстве своём на новичков. Тут мы поговорим о том, как не упереться в лимиты подключений к базе, и чтобы приложение в продакшн не упало.
В этой статье будет идти речь о SQLAlchemy и частности PostgreSQL. В Django таких проблем по умолчанию я не видел, а вот в алхимии, приходится вручную мониторить подключения.
Стандартное создание engine и сессии для алхимии вида:
engine = create_async_engine( SQLALCHEMY_DATABASE_URL, **( dict(pool_recycle=900, pool_size=100, max_overflow=3) ) ) SessionLocal = sessionmaker( autocommit=False, autoflush=False, bind=engine, expire_on_commit=False, class_=AsyncSession )
UPD. Здесь мы будем работать с асинхронной сессией алхимии, но у синхронной такой же механизм действий
И потом мы применяем это примерно так:
async def foo(some: Any): # создаем транзакцию db = SessionLocal() some_do(some) # делаем коммит await db.commit() # закрываем соединение await db.close()
НО, что если на моменте some_do() случится ошибка? Тогда у нас не сделается коммит(ну в принципе логично), и что более важно, не закроется соединение, а это уже критично.
Разберемся почему это критично:
Превысив лимит на подключения к базе, другие запросы не смогут обрабатываться
Лишняя нагрузка на систему
Дополнительные открытые подключения в системе
И если лишняя нагрузка на систему и открытые подключения в системе это хоть и критично, но наше приложение от этого не перестанет работать(до какого то момента). То вот при лимите подключений, уже всё, наше приложение перестанет работать сразу и пользователи будут видеть ошибку
Мы разобрались в чем проблема не закрытых соединений, но как тогда с ними работать?
Можно сделать просто try: … except: …
async def foo(some: Any): # создаем транзакцию db = SessionLocal() try: some_do(some) except Exception as e: print(e) await db.commit() await db.close()
Скажем так, оно будет работать, но всегда есть одно но.. вы серьезно захотите каждый раз делать try except? Всё таки надо найти какой-то более просто способ, чтобы сократить однотипные действия в функции, и убрать вложенность.
Первое что пришло в голову это middlewares, но оно подойдет только если ваше приложение позволяет его реализовать(веб-сервер, телеграм бот), а что если нам нужен универсальный обработчик транзакций? Тут из тени выходит контекстный менеджер
async def foo(some: Any): # создаем транзакцию async with SessionLocal() as s: some_do(some) await db.commit()
Мы определенно сократили количество кода в нашей функции, но всё так же остался лишний уровень вложенности, как мы можем избавиться от него? Передать сессию в функцию из декоратора..? Давайте попробуем
def transaction(): def wrapper(cb): async def wrapped(*args, **kwargs): async with SessionLocal() as session: result = await cb(session, *args, **kwargs) await session.commit() return result return wrapped return wrapper @transaction() async def foo(session: AsyncSession, some: Any): some_do(session, some)
Хмм, кода стало больше, но зато в функции теперь без вложенности и минимальное количество строк. Но погодите-ка, мы каждый раз должны передавать в функцию объект сессии, что создает 1 дополнительную зависимость. Попробуем решить, а как? Просто получать из переменной объект сессии? Да, так в питоне можно, будем использовать contextvars. Можете посмотреть как там что работает, но если кратко, то, что мы обьявляем в родительской функции в переменной ContextVar, то значение передается в вызываемые функции.
Немного примеров:
import asyncio from contextvars import ContextVar var: ContextVar[int] = ContextVar('var') # Тайп хинт показывает, что мы планируем передавать числовой тип в переменную async def bar(): value = var.get() print(f"In context var value is {value}") async def foo(): token = var.set(777) await bar() var.reset(token) asyncio.run(foo()) # In context var value is 777
На выходе мы получаем строку: In context var value is 777
Как видно, мы не передаем явно в функцию никаких значений, но это значение доступно по переменной, и в этом вся магия.
Значит мы можем применить это и к нашей сессии:
P = ParamSpec("P") T = TypeVar("T") def require_session(): session = db_session_var.get() assert session is not None, "Session context is not provided" return session def transaction(): def wrapper( cb: Callable[P, Coroutine[Any, Any, T]] ) -> Callable[P, Coroutine[Any, Any, T]]: @wraps(cb) async def wrapped(*args: P.args, **kwargs: P.kwargs) -> T: if db_session_var.get() is not None: return await cb(*args, **kwargs) async with cast(AsyncSession, SessionLocal()) as session: with use_context_value(db_session_var, session): result = await cb(*args, **kwargs) await session.commit() return result return wrapped return wrapper @contextmanager def use_context_value(context: ContextVar[T], value: T): reset = context.set(value) try: yield finally: context.reset(reset) db_session_var: ContextVar[AsyncSession | None] = ContextVar("db_session_var", default=None)
UPD. Это уже окончательный вид для нашего менеджера, с type hinting и всякими плюшками, чтобы было приятнее работать.
Получается огромный блок для менеджера транзакций, но зато теперь у нас есть простой доступ к нашему объекту сессии:
@transaction() async def foo(): db = require_session() some_do()
И, в чем основная прелесть этого декоратора, то что для под вызываемых функций мы так же их оборачиваем их в transaction(), но у нас используется одна и та же сессия, что дает нам 1 подключение на 1 обработку запроса.
Теперь даже если что-то случится во время выполнения запроса, то у нас автоматически закроется сессия, и у нас не создается очередь из подключений ожидающих закрытие.
Сам этот менеджер-декоратор я использую на своих проектах, он универсальный, будет работать и для тг-ботов, и для веба, да и просто для какого-либо обработчика(главное правильно поставить в порядке декораторов). Мне про этот менеджер-декоратор подсказал мой коллега, за что я ему чрезмерно благодарен.
Спасибо за то что прочитали статью! Если у вас, опытных разработчиков, есть какие-то замечания к статье, буду ждать в комментариях вас
