Асинхронні контекстні менеджери

9.1 Контекстні менеджери

Ви пам'ятаєте, як ми використовували оператор with, щоб контролювати доступ до ресурсу? З його допомогою можна було точно відстежити, коли ресурс почав використовуватись, і коли цей процес закінчився. Те ж саме можна робити і для «асинхронних ресурсів».

Асинхронні контекстні менеджери (AsyncContextManager) використовуються для управління ресурсами в асинхронному програмуванні, забезпечуючи правильне відкриття і закриття ресурсів в асинхронних функціях. Вони працюють аналогічно звичайним контекстним менеджерам, але призначені для використання в асинхронних функціях з ключовими словами async with.

Я не буду детально про це розповідати, бо найближчим часом ви точно не будете створювати свої асинхронні контекстні менеджери, а до того часу, коли вирішите це робити, все одно все забудете. Але ознайомити із самою концепцією хочу.

Створення асинхронного контекстного менеджера

Асинхронні контекстні менеджери визначаються з використанням методів __aenter__ і __aexit__. Ці методи аналогічні методам __enter__ і __exit__ в звичайних контекстних менеджерах, але є асинхронними.

• __aenter__(): Асинхронний метод, що викликається при вході в контекст.
• __aexit__(exc_type, exc, tb): Асинхронний метод, що викликається при виході з контексту. Приймає три аргументи: тип виключення, саме виключення і трасування (traceback).

Приклад створення асинхронного контекстного менеджера

import asyncio

class AsyncContextManager:
    async def __aenter__(self):
        print("Enter context")
        return self
        
    async def __aexit__(self, exc_type, exc, tb):
        print("Exit context")
        
async def main():
    async with AsyncContextManager():
        print("Inside context")
        
asyncio.run(main())

Ось як буде виконуватись цей код:

1. Асинхронно запуститься функція main()
2. Функція main() розпочне виконання
3. Створиться об'єкт типу AsyncContextManager
4. У об'єкта AsyncContextManager викличеться метод __aenter__()
5. Виведеться print("Enter context")
6. Виконається код print("Inside context")
7. У об'єкта AsyncContextManager викличеться метод __aexit__()
8. Завершиться виконання функції main()

9.2 Приклади використання

У прикладі вище не обов'язково використовувати саме асинхронну версію менеджера контексту — код буде працювати, навіть якщо його зробити синхронним.

Але асинхронний менеджер потрібен, якщо всередині блоку with викликається асинхронна функція або використовується оператор await.

Асинхронний контекстний менеджер для роботи з файлами

У цьому прикладі використовується бібліотека aiofiles для асинхронного читання і запису файлів. Асинхронний контекстний менеджер aiofiles.open дозволяє безпечно відкривати і закривати файли в асинхронному контексті.

import aiofiles
import asyncio
            
async def main():
    async with aiofiles.open('example.txt', mode='w') as file:
        await file.write('Hello, world!')
            
asyncio.run(main())

Асинхронний контекстний менеджер для мережевих операцій

У цьому прикладі використовується бібліотека aiohttp для виконання асинхронних HTTP-запитів. Асинхронні контекстні менеджери ClientSession та session.get забезпечують правильне управління з'єднаннями.

import aiohttp
import asyncio
            
async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()
            
async def main():
    html = await fetch('https://example.com')
    print(html)
            
asyncio.run(main())

Асинхронні контекстні менеджери забезпечують автоматичне управління ресурсами, такими як файли, мережеві з'єднання та інші ресурси, які необхідно відкривати і закривати.

Методи __aenter__ і __aexit__ дозволяють виконувати асинхронні операції при вході і виході з контексту, забезпечуючи паралельне виконання завдань. Використання асинхронних контекстних менеджерів допомагає уникнути витоків ресурсів і гарантує, що всі ресурси будуть коректно звільнені.