Контекстные менеджеры Python: создаём, используем и не наступаем на грабли

Каждый разработчик, проработавший с Python больше пары месяцев, неизбежно сталкивался с ситуацией, когда открытый файл почему-то «забыл» закрыться, соединение с базой данных повисло, а ресурсы оперативной памяти «убежали» в неизвестном направлении. Если вы относитесь к счастливчикам, которым всё ещё незнакомы эти проблемы, поздравляю, вы уникальны (и, вероятно, пишете исключительно «Hello World!»). Всем остальным приходится постоянно задумываться о надёжном и безопасном использовании ресурсов, будь то файлы, сетевые подключения, сессии с базами данных или даже системные процессы.

Именно здесь на помощь приходит контекстный менеджер — особая конструкция в Python, обеспечивающая удобную и безопасную работу с ресурсами, избавляя вас от рутинного ручного контроля их жизненного цикла. Используя конструкцию with, вы будто говорите Python: «Сделай за меня скучную работу и закрой всё, что я забыл закрыть». Поверьте, ваша память скажет вам за это спасибо (особенно если её всего пара гигабайт, а вы любите открывать сразу сотню файлов).

Но как именно работает эта магия «под капотом»? На самом деле, всё довольно просто и логично. Когда вы пишете такой код:

with open('file.txt', 'r') as file:
    data = file.read()

происходит следующее: Python вызывает специальный метод __enter__() объекта (в нашем случае файла), который и возвращает сам объект для дальнейших действий. После выполнения блока кода Python автоматически вызывает метод __exit__(), который заботится о безопасном закрытии файла, даже если вы его каким-то чудом забыли закрыть сами (забыли, конечно, совершенно случайно).

Эта простая модель гарантирует, что ресурсы освобождаются ровно в тот момент, когда это нужно, не дожидаясь, пока сборщик мусора «доберётся» до них в неопределённом будущем. Такой подход минимизирует риск утечки памяти, экономит ресурсы системы и упрощает отладку. Причём касается это не только файлов, но и любых других внешних ресурсов: баз данных, сетевых соединений, потоков и многого другого.

Но на этом приятности контекстных менеджеров не заканчиваются. Помимо удобного управления ресурсами, конструкция with заметно улучшает читаемость и логичность кода. Любой, кто позже откроет ваш код (включая вас самих через полгода), сразу поймёт: здесь используется ресурс, и он гарантированно освободится. Без необходимости погружаться в десятки строк ручного управления, условий и обработок исключений, вы получаете ясную, лаконичную и надёжную структуру.

В следующих главах мы подробно разберём, как создавать собственные синхронные и асинхронные контекстные менеджеры, используя возможности самых популярных современных библиотек, и как избежать распространённых ошибок. Мы рассмотрим реальные сценарии и увидим, как правильное использование конструкции with упрощает жизнь разработчика и помогает сохранить разум ясным, код читаемым, а сервер живым. Поверьте, это стоит того, чтобы потратить на изучение несколько минут вашего драгоценного времени.

Хотите откровений? Пожалуйста: большинство библиотек, которые вы используете ежедневно, буквально держатся на контекстных менеджерах. Это именно тот случай, когда простой синтаксис решает сложные проблемы за вас. Чтобы вы поняли, насколько далеко зашло это удобство, давайте без лишних абстракций рассмотрим конкретные примеры на самых актуальных версиях популярных Python-библиотек. Не волнуйтесь, после прочтения этой главы вы будете использовать конструкцию with так уверенно, словно родились с клавиатурой в руках.

Начнём, пожалуй, с классики жанра — SQLAlchemy 2.x. Эта библиотека незаменима в работе с базами данных и, в частности, при взаимодействии с транзакциями и сессиями. Если вы до сих пор вручную открываете и закрываете сессии, мой вам совет: прекратите. Используя контекстный менеджер, вы экономите себе массу времени и нервов. Сравните:

Плохо (классический, «ручной» подход):

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql+psycopg://user:pass@localhost/db')
Session = sessionmaker(bind=engine)
session = Session()
try:
    session.execute(...)  # какая-то очень важная операция
    session.commit()
except Exception:
    session.rollback()
    raise finally:
    session.close()

Идеально (используем магию контекстного менеджера):

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql+psycopg://user:pass@localhost/db')
Session = sessionmaker(bind=engine)

with Session() as session:
    session.execute(...)  # теперь это безопасно и просто
    session.commit()

Красиво, правда? Благодаря использованию контекстного менеджера вам больше не нужно волноваться о ручном закрытии сессий и откате транзакций. Библиотека сама корректно завершит работу с базой данных, даже если в вашем коде неожиданно появится ошибка (а ошибки, как известно, имеют склонность неожиданно появляться).

Перейдём к следующему герою сегодняшнего выпуска — Requests 2.32.x. Эту библиотеку разработчики используют чаще, чем собственный смартфон. Но и здесь встречаются досадные ошибки, связанные с «зависшими» HTTP-соединениями, которые долго висят в ожидании закрытия. Решение снова на поверхности — контекстный менеджер:

Код без контекстного менеджера, способный вызвать головную боль:

import requests

response = requests.get('https://example.com/api')
data = response.json()
response.close()  # так легко забыть эту строчку 

Элегантное решение:

import requests

with requests.get('https://example.com/api') as response:
    data = response.json()

Теперь Requests автоматически закроет соединение сразу после завершения блока. Никаких утечек памяти, никаких забытых соединений, никаких ночных звонков от администратора, которого замучили сообщения о превышении лимита открытых сокетов.

Наконец, поговорим о Psycopg 3.2.x — современном драйвере для PostgreSQL. И здесь без контекстных менеджеров далеко не уйти (точнее уйти можно, но возвращаться придётся очень быстро). Вместо привычного «открыл-закрыл» вручную, можно использовать конструкцию with, которая гарантирует стабильную и безопасную работу с базой:

Сравним классический способ с более правильным подходом:

Старомодный и опасный подход:

import psycopg

conn = psycopg.connect("dbname=db user=user password=pass")
cur = conn.cursor()
try:
    cur.execute("SELECT * FROM table;")
    rows = cur.fetchall()
finally:
    cur.close()
    conn.close()

Безопасный и современный способ (вы уже наверняка догадались, какой):

import psycopg

with psycopg.connect("dbname=db user=user password=pass") as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM table;")
        rows = cur.fetchall()

Благодаря контекстному менеджеру Psycopg сам позаботится о том, чтобы курсор и соединение были корректно закрыты, даже если что-то пойдёт не так. А поверьте, в продакшене рано или поздно что-то обязательно пойдёт не так.

Однако не стоит забывать и о типичных ошибках, которые могут возникнуть при неправильном использовании контекстных менеджеров. Например, одна из распространённых — это попытка использовать закрытый ресурс после выхода из блока with. Запомните простое правило: если вы вышли из блока контекстного менеджера, ресурс уже закрыт, и пытаться использовать его снова не стоит — это как пытаться попить чай из пустой чашки (интересно, но бесполезно).

Таким образом, синхронные контекстные менеджеры — это не просто удобство. Это стандарт хорошего тона, экономия времени и ресурсов, и ваш гарант того, что однажды ночью вас не разбудит админ с вопросом: «Ты точно закрыл соединения?». В следующей главе мы обсудим ещё более увлекательную тему — асинхронные контекстные менеджеры, без которых вообще невозможно представить современную асинхронную разработку. Поверьте, там вас ждёт много интересного.

Когда-то давно, в далёкие времена, асинхронное программирование в Python было уделом энтузиастов, которые любили «придумывать себе проблемы». Сегодня же асинхронность стала реальностью, частью повседневной разработки, а библиотека asyncio — таким же привычным инструментом, как зубная щётка. И если вы уже окунулись в мир async/await, то наверняка знаете: без грамотного управления ресурсами асинхронный код становится лабиринтом, из которого почти невозможно выбраться без потерь — будь то утечки памяти или оставшиеся висеть соединения.

Асинхронные контекстные менеджеры как раз и решают эти проблемы, предоставляя нам удобный способ контролировать жизненный цикл ресурсов в async-коде. Казалось бы, что может пойти не так, если вы просто забыли закрыть асинхронное соединение? На самом деле, очень многое: начиная от бесконечно висящих открытых подключений и заканчивая необъяснимыми таймаутами, которые сведут с ума даже самых опытных разработчиков.

Рассмотрим всё это на практических примерах. Начнём с одной из самых востребованных сегодня библиотек — асинхронного HTTP-клиента httpx 0.27.x. Типичная ошибка начинающих async-разработчиков — открытие клиентской сессии без последующего закрытия. В результате приложение рискует столкнуться с неожиданной нехваткой ресурсов уже через пару часов работы:

Ошибочный подход (не повторяйте дома):

import httpx
import asyncio

async def fetch_data():
    client = httpx.AsyncClient()
    response = await client.get('https://example.com/api')
    data = response.json()
    return data

А вот правильный способ, с применением асинхронного контекстного менеджера:

import httpx
import asyncio

async def fetch_data():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://example.com/api')
        return response.json()

Красота, простота и безопасность — и всё это благодаря конструкции async with. Клиентская сессия закроется сама, даже если где-то возникнет неожиданная ошибка.

Продолжим нашу прогулку по миру async-библиотек и обратимся к aioredis-py 2.x, которая теперь встроена прямо в redis-py и используется для взаимодействия с Redis. Даже тут контекстный менеджер играет важнейшую роль, предотвращая подвисания соединений и накопление «мусора»:

Распространённый, но неудачный подход:

import redis.asyncio as aioredis

async def cache_data():
    redis_client = aioredis.Redis(host='localhost', port=6379)
    await redis_client.set('key', 'value')
    await redis_client.close()

Лучший подход — «один раз сделал и забыл»:

import redis.asyncio as aioredis

async def cache_data():
    async with aioredis.Redis(host='localhost', port=6379) as redis_client:
        await redis_client.set('key', 'value')

Больше никаких случайных забытых вызовов close(), никаких мучительных отладок и поисков причин внезапных падений производительности.

Теперь поговорим об asyncpg 0.29.x — одном из самых эффективных и популярных асинхронных драйверов для PostgreSQL. Работа с базами данных в async-мире особенно чувствительна к забытым ресурсам: каждое подвисшее подключение — это потенциальная «бомба замедленного действия» для вашего приложения. Поэтому без контекстного менеджера здесь лучше не экспериментировать вовсе:

Плохой подход («авось пронесёт»):

import asyncpg

async def fetch_records():
    conn = await asyncpg.connect(user='user', password='pass', database='db', host='localhost')
    records = await conn.fetch('SELECT * FROM table;')
    await conn.close()
    return records

Правильный, понятный и безопасный подход:

import asyncpg

async def fetch_records():
    async with asyncpg.connect(user='user', password='pass', database='db', host='localhost') as conn:
        return await conn.fetch('SELECT * FROM table;')

И снова «магия» контекстного менеджера не только закрывает соединения вовремя, но и делает код чище, а вас — немного счастливее.

Как видно, асинхронные контекстные менеджеры — это далеко не прихоть ленивых разработчиков, а насущная необходимость при работе с любой серьёзной библиотекой, которая асинхронно общается с внешними ресурсами. Они гарантируют, что ресурсы освобождаются вовремя, приложение не страдает от утечек памяти, а ваш сон не прерывают уведомления об ошибках в три часа ночи.

Подводя итог главы, напомню главное: если вы работаете с async-кодом в Python, контекстные менеджеры — это ваше всё. Используйте их везде, где можете, и не придётся краснеть перед коллегами за внезапно повисшее приложение. А в следующей части мы поговорим о типичных ошибках, которые допускают даже профессионалы при создании и использовании контекстных менеджеров, и о том, как сделать так, чтобы вы сами не стали героем печальных анекдотов в рабочем чате.

Как показывает практика, даже опытные Python-разработчики не застрахованы от ошибок при написании собственных контекстных менеджеров. Казалось бы, конструкция простая и очевидная — бери и используй. Но не спешите расслабляться: именно в кажущейся простоте и кроются коварные ловушки. Сейчас мы разберёмся, как создать свой контекстный менеджер и не стать при этом «героем» пятничного разбора в вашей команде.

Первая и любимая ошибка начинающих «творцов» контекстных менеджеров — неправильная работа с исключениями. Казалось бы, вы аккуратно написали __enter__() и __exit__(), но вдруг обнаруживаете, что ресурсы упорно отказываются закрываться при появлении исключений. Вот типичный пример ошибочного подхода:

class FileManager:
    def __init__(self, file_name, mode):
        self.file_name = file_name
        self.mode = mode

    def __enter__(self):
        self.file = open(self.file_name, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()

Казалось бы, что не так? Но если открыть файл не удалось (например, файл не существует или нет прав доступа), метод __enter__() выбросит исключение, а метод __exit__() вызван не будет. Результат? Утечка ресурсов и память, уплывающая в закат. Чтобы избежать подобной ситуации, следует инициализировать ресурс заранее и безопасно обрабатывать исключения, например:

class FileManager:
    def __init__(self, file_name, mode):
        self.file_name = file_name
        self.mode = mode
        self.file = None

    def __enter__(self):
        try:
            self.file = open(self.file_name, self.mode)
            return self.file
        except Exception as e:
            print(f'Не удалось открыть файл: {e}')
            raise

    def __exit__(self, exc_type, exc_val, exc_tb):
        if hasattr(self, 'file') and not self.file.closed:
            self.file.close()

Таким образом, ресурс гарантированно будет закрыт, даже если «всё пойдёт не по плану». Это уже похоже на профессиональный подход, а не на «пальцем в небо».

Вторая распространённая ошибка — забывать, что метод __exit__() может и должен реагировать на исключения. Представьте, вы написали код, который автоматически коммитит транзакцию при выходе из контекста базы данных. А что, если произошла ошибка? Вы ведь не хотите сохранить битые данные, верно? Корректная обработка исключений должна выглядеть примерно так:

class DBTransaction:
    def __init__(self, session):
        self.session = session

    def __enter__(self):
        return self.session

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.session.commit()
        else:
            self.session.rollback()
            print(f'Транзакция отменена из-за ошибки: {exc_val}')

Обратите внимание, что если в коде возникнет исключение, транзакция автоматически откатится, сохраняя целостность ваших данных. Забудете об этом, и вместо вечернего кофе будете искать способ откатить несколько сотен ошибочных записей в продакшене.

Ещё одна типичная ошибка — избыточная автоматизация и переусложнение контекстного менеджера. Порой программисты настолько увлекаются «магией», что создают контекстные менеджеры, которые делают абсолютно всё (включая варку кофе и отправку смс начальнику). Помните: излишняя автоматизация порождает ошибки там, где их могло и не быть. Контекстный менеджер должен выполнять ровно одну задачу — управление ресурсом. Всё остальное — избыточная сложность, создающая новые точки отказа.

Ещё один подводный камень связан с асинхронными контекстными менеджерами. Многие путают методы __enter__/__exit__ и __aenter__()/__aexit__(). Попытка использовать синхронный менеджер в async-коде без адаптации приведёт к непредсказуемым последствиям, и вот вы уже героически боретесь с event loop’ом, который замер в ожидании чуда.

Пример правильной реализации асинхронного менеджера:

class AsyncDBConnection:
    async def __aenter__(self):
        self.conn = await asyncpg.connect(user='user', password='pass')
        return self.conn

    async def __exit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

Именно асинхронные версии __aenter__() и __aexit__() должны использоваться с ключевым словом async with.

Ещё один важный момент — контекстные менеджеры не являются панацеей от всех ошибок. Они лишь гарантируют освобождение ресурсов, но если вы забыли обработать важные исключения внутри самого контекста, ваш код будет выглядеть красиво, но работать — ужасно. Поэтому старайтесь совмещать надёжность автоматического закрытия ресурсов с грамотной обработкой ошибок в самом коде бизнес-логики.

Подытоживая эту «парадоксальную» главу, важно подчеркнуть простую мысль: контекстный менеджер — ваш верный помощник, но не личный телохранитель. Он поможет избежать типичных ошибок и сэкономить вам часы отладки, но не спасёт от логических просчётов и чрезмерной самоуверенности. Используйте его мудро и по назначению, а в следующей главе мы с вами создадим универсальный контекстный менеджер, который идеально работает как синхронно, так и асинхронно, чтобы сделать вашу жизнь ещё проще и немного приятнее.

Если вы когда-нибудь задумывались о том, почему в Python существует два отдельных типа контекстных менеджеров (синхронные и асинхронные), и нельзя ли сделать один универсальный «менеджер всего на свете», значит вы настоящий прагматик. К счастью, в мире Python нет ничего невозможного — кроме разве что простого перехода на Python 2.7 обратно (ну и ещё запуска TensorFlow на калькуляторе, хотя кто знает).

Итак, настало время взять в свои руки реализацию универсального контекстного менеджера, который одинаково комфортно чувствует себя и в синхронном, и в асинхронном коде. Такая конструкция востребована на практике, особенно когда код должен гибко переключаться между синхронными и асинхронными библиотеками (например, при работе с файловыми системами, кешами или удалёнными сервисами).

Начнём с основы, а именно — определения структуры класса. Универсальный контекстный менеджер требует реализации сразу четырёх методов: синхронных (__enter__ и __exit__) и асинхронных (__aenter__ и __aexit__). На простом примере, допустим, хотим реализовать менеджер для подключения к какому-то воображаемому ресурсу (например, сетевому соединению), поддерживающему оба подхода:

import asyncio

class UniversalResourceManager:
    def __init__(self, resource_name):
        self.resource_name = resource_name
        self.resource = None

    # Синхронное открытие ресурса
    def __enter__(self):
        print(f"[SYNC] Открываем ресурс {self.resource_name}")
        self.resource = self._open_resource()
        return self.resource

    # Синхронное закрытие ресурса с обработкой исключений
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type := exc_val:
            print(f"[SYNC] Возникла ошибка: {exc_val}")
        print(f"[SYNC] Закрываем ресурс {self.resource_name}")
        self._close_resource()

    # Асинхронное открытие ресурса
    async def __aenter__(self):
        print(f"[ASYNC] Асинхронно открываем ресурс {self.resource_name}")
        self.resource = await self._async_open_resource()
        return self.resource

    # Асинхронное закрытие ресурса
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_val := exc_val:
            print(f"[ASYNC] Произошла ошибка: {exc_val}")
        print(f"[ASYNC] Асинхронно закрываем ресурс {self.resource_name}")
        await self._async_close_resource()

    # Приватные методы (для примера, простейшие реализации)
    def _open_resource(self):
        self.resource = open(self.resource_name, 'w')  # Допустим, обычный файл

    def _close_resource(self):
        if self.resource and not self.resource.closed:
            self.resource.close()

    async def _async_open_resource(self):
        await asyncio.sleep(1)  # имитация асинхронного открытия
        self.resource = f"AsyncResource({self.resource_name})"
        return self.resource

    async def _async_close_resource(self):
        await asyncio.sleep(0.1)
        self.resource = None

Посмотрите внимательно: мы разделили логику работы с ресурсами на синхронную и асинхронную части, что позволяет использовать менеджер в любом сценарии. Теперь использование нашего менеджера выглядит так просто и удобно, что даже непонятно, зачем когда-то было иначе:

Синхронный пример
with UniversalResourceManager("data.txt") as resource:
    print("[SYNC] Работаем с ресурсом внутри блока")

Асинхронный вариант (идеально для веб-сервисов на FastAPI или Async Django):

import asyncio

async def main():
    async with UniversalManager("remote_service") as resource:
        print(f"Используем ресурс {resource} внутри async-блока")

asyncio.run(main())

Особенность нашей реализации в том, что мы чётко разделили управление ресурсом для синхронного и асинхронного подходов, что обеспечивает максимальную безопасность и прозрачность работы. Заметьте: никаких пересечений методов быть не должно! Синхронные методы никогда не должны «лезть» в асинхронную логику и наоборот — иначе вы легко можете запутать event loop или заблокировать всю систему. А поверьте, нет ничего более грустного, чем зависшее асинхронное приложение, которое заблокировало само себя.

Кроме того, обратите внимание на обработку исключений: метод __exit__ принимает три параметра (exc_type, exc_val, exc_tb) и позволяет грамотно реагировать на ошибки, решая, откатить ли транзакцию, отменить изменения или просто «проглотить» исключение. В асинхронной версии — всё то же самое, только с приставкой async, и «проглотить» исключение так же легко, как и в синхронной. Главное, помните — если вы игнорируете исключения, будьте готовы потом объяснять, почему сервис падает молча (а он обязательно упадёт).

Итак, теперь в вашем арсенале есть удобный и универсальный инструмент, способный закрыть за вами все забытые соединения, файлы и прочие ресурсы. Это не просто удобно — это наглядный пример того, как правильно построенный контекстный менеджер избавляет вас от лишних телодвижений, экономит ресурсы и нервы (ваши и админов, которые теперь реже будут звонить с претензиями).

В следующей главе мы займёмся тем, что окончательно и бесповоротно разберём по «винтикам» внутренности нашего менеджера, оценим его реальные перспективы в продакшене и порассуждаем, стоит ли вообще городить велосипед, когда вокруг и так полно решений, которые кто-то уже придумал до нас. Но если уж и создавать свой велосипед, то пусть он будет хотя бы с мягким сиденьем и педалями — и это мы с вами точно обеспечим.

Если вы добрались до этой главы и не бросили свой новый контекстный менеджер на полпути, поздравляю — вы один из немногих. Теперь пора заглянуть под капот и понять, почему же этот инструмент работает именно так, как задумано, и какие секреты скрывает за кажущейся простотой.

На первый взгляд, реализация универсального контекстного менеджера выглядит тривиально. Всего несколько магических методов (__enter__, __exit__, __aenter__, __aexit__) — и готово, можно расходиться. Однако на деле эти методы не такие уж простые. Начнём с синхронных методов.

Метод __enter__() не просто инициализирует ресурс, но и должен чётко сигнализировать пользователю (вашему коду), что ресурс готов к использованию. Самая частая ошибка — возвращать ресурс, не проверив его готовность. Правильный подход — предварительно убедиться, что ресурс был успешно создан, иначе вы рискуете использовать нерабочий объект дальше по коду. Вот пример правильной проверки:

def __enter__(self):
    try:
        self.resource = open(self.resource_name, self.mode)
        return self.resource
    except IOError as e:
        print(f"Ошибка открытия ресурса: {e}")
        raise 

Аналогично, метод __exit__() должен не просто закрывать ресурс, а обрабатывать исключения, возникающие внутри блока. Здесь важно помнить, что метод принимает три аргумента: тип исключения, его экземпляр и объект traceback. Корректное использование метода позволит вам, например, отменить изменения, если что-то пошло не так, или, наоборот, «проглотить» исключение, если оно вам кажется незначительным:

def __exit__(self, exc_type, exc_val, exc_tb):
    if exc_type:
        print(f"Исключение внутри контекста: {exc_val}. Откатываем изменения.")
        self.rollback_resource()
    else:
        self.commit_resource()
    self.close_resource()

Асинхронные версии этих методов, __aenter__() и __aexit__(), требуют особой аккуратности. Они работают внутри event loop’а и обязаны использовать асинхронные вызовы (await). Малейшее отступление от асинхронности — и вся ваша carefully-crafted async-архитектура рискует превратиться в хаос.

Например, __aenter__() должен быть написан так, чтобы не блокировать event loop:

async def __aenter__(self):
    try:
        self.resource = await self.async_open_resource()
        return self.resource
    except Exception as e:
        print(f"Ошибка при асинхронном открытии ресурса: {e}")
        raise 

А вот __aexit__() заботится о корректном освобождении ресурса с учётом возможного исключения в async-коде:

async def __aexit__(self, exc_type, exc_val, exc_tb):
    if exc_type:
        await self.async_rollback()
        print(f"Произошла ошибка: {exc_val}. Выполнен откат.")
    else:
        await self.async_commit()
    await self.async_close_resource()

Теперь немного поговорим о потенциальных улучшениях. Куда расти дальше? Конечно же, нет предела совершенству: можно встроить логирование всех операций, добавить обработку специфичных исключений, встроить таймауты для длительных операций (очень советую!), а также предусмотреть возможность переиспользования ресурсов через пул соединений (connection pool).

К слову, использование connection pooling — один из важнейших способов улучшения производительности в высоконагруженных системах. Вместо постоянного создания и закрытия соединений, вы берёте их из заранее созданного пула и возвращаете обратно после использования. Таким образом, вместо лишней нагрузки и задержек вы получаете стабильность и мгновенную доступность ресурсов.

Вот простая схема, как улучшить наш менеджер с пулом:

from contextlib import asynccontextmanager
import asyncpg

class AsyncDBPoolManager:
    def __init__(self, dsn):
        self.dsn = dsn
        self.pool = None

    async def init_pool(self):
        self.pool = await asyncpg.create_pool(dsn=self.dsn)

    async def close_pool(self):
        await self.pool.close()

    async def __aenter__(self):
        if not self.pool:
            await self.init_pool()
        self.conn = await self.pool.acquire()
        return self.pool

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.pool.release(self.pool)


# Пример использования: async def main():
manager = DBPoolManager(dsn='postgresql://user:pass@localhost/db')
await manager.init_pool()
async with manager.pool.acquire() as conn:
    await conn.execute('SELECT * FROM table;')

await manager.close_pool()

Этот подход — «золотой стандарт» в крупных production-системах, и ваш контекстный менеджер теперь не просто надёжен, но и полностью готов к реальному применению в условиях серьёзных нагрузок.

И, конечно, немного полезного цинизма напоследок: прежде чем кидаться создавать собственный контекстный менеджер, задумайтесь — а может, кто-то уже давно решил вашу проблему? Пишите своё решение, только если оно действительно будет удобнее и эффективнее стандартного, иначе вы рискуете изобрести очередной велосипед. И всё же, если велосипед уже появился, позаботьтесь хотя бы о том, чтобы он ехал, а не катился по инерции под гору. Контекстные менеджеры — это именно тот случай, когда правильная реализация действительно оправдывает потраченное время и экономит вам десятки часов будущей отладки.

Теперь, разобравшись во всех деталях и тонкостях, вы готовы применить полученные знания и уверенно управлять ресурсами, которые доверите своему новому, удобному и безопасному контекстному менеджеру. И помните: хороший код — это тот, который работает, когда вы уже забыли, что его писали именно вы.