Когда вы пишете асинхронный код, становится гораздо проще запускать несколько операций одновременно. Но из-за того, что всё крутится вокруг событийного цикла (event loop), обработка ошибок может немного усложниться.
Представьте, что у вас одновременно выполняются несколько задач. Если одна из них рухнула с ошибкой, это не значит, что нужно останавливать все остальные. Более того, если не обрабатывать исключения аккуратно, они могут просто "пропасть" — особенно если вы забыли про await или не обернули вызов в try-except.
Вроде бы в асинхронном коде мы делаем почти то же самое, что и в обычном — ловим исключения, проверяем ошибки. Но из-за параллельной природы задач нужно быть внимательнее: одна ошибка может пролететь мимо, пока вы заняты чем-то другим. Так что обрабатывать их нужно с умом — и желательно заранее подумать, где и как именно.
Асинхронная обработка исключений: основные принципы
Чтобы понять, как обрабатывать исключения, возникающие в асинхронном коде, давайте рассмотрим несколько ключевых принципов:
tryиexceptработают в асинхронных функциях:
асинхронные функции поддерживают стандартный Python-синтаксис обработки исключений. Вы можете использоватьtry/exceptвнутри функций сasync def.- Обработка исключений в задачах:
когда вы создаете асинхронные задачи с помощьюasyncio.create_task, важно учитывать, что исключения из таких задач не всплывают автоматически. Их нужно обрабатывать вручную. - Управление контекстом ошибок:
важно иметь глобальный механизм для отлова исключений, которые могут остаться необработанными. FastAPI предоставляет возможность задавать глобальные обработчики ошибок.
Пример обработки ошибок в асинхронных функциях
Начнем с самого простого: как обрабатывать исключения внутри одной асинхронной функции.
from fastapi import FastAPI
from asyncio import sleep
app = FastAPI()
@app.get("/example")
async def example_endpoint():
try:
# Симуляция исключения через sleep
await sleep(1)
raise ValueError("Упс! Что-то пошло не так.")
except ValueError as e:
# Обрабатываем исключение и возвращаем корректный ответ
return {"error": str(e)}
Здесь мы используем try и except так же, как в обычном коде. Если при вызове /example возникнет ошибка, клиент получит ее описание в ответе.
Как обрабатывать ошибки при создании задач
Если мы работаем с несколькими задачами, обработка ошибок становится сложнее. Рассмотрим пример, где мы создаем несколько задач сразу:
import asyncio
from fastapi import FastAPI
app = FastAPI()
async def risky_task(task_id: int):
# Имитация задачи, которая может выбросить ошибку
if task_id % 2 == 0:
raise RuntimeError(f"Ошибка в задаче {task_id}")
return f"Задача {task_id} выполнена успешно"
@app.get("/tasks")
async def run_tasks():
tasks = [asyncio.create_task(risky_task(i)) for i in range(5)]
results = []
for task in tasks:
try:
result = await task
results.append(result)
except RuntimeError as e:
# Ловим ошибку из задачи и обрабатываем
results.append({"error": str(e)})
return {"results": results}
Мы создаем 5 задач, где каждая вторая задача выбрасывает исключение. Задачи выполняются параллельно, а их ошибки обрабатываются в цикле for.
Использование asyncio.gather для обработки ошибок
Когда вы вызываете несколько задач одновременно, удобно использовать функцию asyncio.gather. Однако, по умолчанию она останавливает выполнение при первом же исключении. Чтобы этого избежать, можно передать аргумент return_exceptions=True.
import asyncio
from fastapi import FastAPI
app = FastAPI()
async def risky_task(task_id: int):
if task_id % 2 == 0:
raise RuntimeError(f"Ошибка в задаче {task_id}")
return f"Задача {task_id} выполнена успешно"
@app.get("/gather-tasks")
async def run_tasks_with_gather():
tasks = [risky_task(i) for i in range(5)]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Обработка результатов
final_results = []
for result in results:
if isinstance(result, Exception):
final_results.append({"error": str(result)})
else:
final_results.append({"success": result})
return {"results": final_results}
С помощью asyncio.gather мы выполняем задачи параллельно, но указываем, что хотим продолжать выполнение даже при возникновении исключений.
Обработка ошибок в асинхронных эндпоинтах FastAPI
Теперь давайте посмотрим, как обработать исключения в асинхронных эндпоинтах FastAPI. Представьте, что у нас есть функция, взаимодействующая с внешним API, где возможны тайм-ауты:
from fastapi import FastAPI
import httpx
app = FastAPI()
async def fetch_data():
async with httpx.AsyncClient() as client:
try:
response = await client.get("https://example.com/data", timeout=5.0)
response.raise_for_status()
return response.json()
except httpx.RequestError as exc:
raise RuntimeError(f"Ошибка при запросе: {exc}")
except httpx.HTTPStatusError as exc:
raise RuntimeError(f"Недопустимый HTTP-ответ: {exc.response.status_code}")
@app.get("/fetch-data")
async def get_data():
try:
data = await fetch_data()
return {"data": data}
except RuntimeError as e:
return {"error": str(e)}
Здесь мы обрабатываем ошибки HTTP-запросов: сетевые проблемы, тайм-ауты и недопустимые статусы HTTP.
Особенности работы с асинхронными исключениями в FastAPI
FastAPI автоматически обрабатывает необработанные исключения и возвращает их в виде HTTP-ответов. Например, если мы случайно пропустим обработку исключения, FastAPI вернет статус 500 и краткое описание ошибки. Однако рекомендуется явно задавать логику обработки ошибок, чтобы сделать приложение более предсказуемым.
Советы новичкам
- Логируйте все неожиданные ошибки:
используйте встроенный механизм логирования FastAPI или популярные библиотеки для логирования, такие какloguru. - Не забывайте про тайм-ауты:
асинхронные задачи могут "зависать". Используйте тайм-ауты для управления длительностью выполнения задач. - Тщательно тестируйте асинхронный код:
асинхронные ошибки труднее отлаживать. Убедитесь, что у вас есть тесты, проверяющие обработку всех важных сценариев.
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ