Асинхронна обробка помилок: як це працює

Коли ти пишеш асинхронний код, набагато простіше запускати кілька операцій одночасно. Але через те, що все крутиться навколо event loop, обробка помилок може трохи ускладнитися.

Уяви, що в тебе одночасно виконуються кілька задач. Якщо одна з них впала з помилкою, це не означає, що треба зупиняти всі інші. Більше того, якщо не обробляти виключення акуратно, вони можуть просто "зникнути" — особливо якщо ти забув про await або не обернув виклик в try-except.

Здається, в асинхронному коді ми робимо майже те саме, що й у звичайному — ловимо виключення, перевіряємо помилки. Але через паралельну природу задач треба бути уважнішим: одна помилка може пролетіти повз, поки ти зайнятий чимось іншим. Тому обробляти їх треба з розумом — і бажано заздалегідь подумати, де й як саме.

Асинхронна обробка виключень: основні принципи

Щоб зрозуміти, як обробляти виключення, що виникають в асинхронному коді, давайте розглянемо кілька ключових принципів:

1. try і except працюють в асинхронних функціях:
   асинхронні функції підтримують стандартний Python-синтаксис обробки виключень. Ти можеш використовувати try/except всередині функцій з async def.
2. Обробка виключень в задачах:
   коли ти створюєш асинхронні задачі за допомогою asyncio.create_task, важливо врахувати, що виключення з таких задач не спливають автоматично. Їх потрібно обробляти вручну.
3. Управління контекстом помилок:
   важливо мати глобальний механізм для відлову виключень, які можуть залишитися необробленими. 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 і короткий опис помилки. Однак рекомендується явно задавати логіку обробки помилок, щоб зробити застосунок більш передбачуваним.

Поради новачкам

1. Логуйте всі несподівані помилки:
   використовуй вбудований механізм логування FastAPI або популярні бібліотеки для логування, такі як loguru.
2. Не забувай про тайм-аути:
   асинхронні задачі можуть "зависати". Використовуй тайм-аути для управління тривалістю виконання задач.
3. Ретельно тестуй асинхронний код:
   асинхронні помилки важче відлагоджувати. Переконайся, що в тебе є тести, які перевіряють обробку всіх важливих сценаріїв.