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())
Ось як буде виконуватись цей код:
- Асинхронно запуститься функція
main()
- Функція
main()
розпочне виконання - Створиться об'єкт типу
AsyncContextManager
- У об'єкта
AsyncContextManager
викличеться метод__aenter__()
- Виведеться
print("Enter context")
- Виконається код
print("Inside context")
- У об'єкта
AsyncContextManager
викличеться метод__aexit__()
- Завершиться виконання функції
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__
дозволяють виконувати асинхронні операції при
вході і виході з контексту, забезпечуючи паралельне виконання завдань.
Використання асинхронних контекстних менеджерів допомагає уникнути витоків
ресурсів і гарантує, що всі ресурси будуть коректно звільнені.