Асинхронность в FastAPI: async/await

Привет! FastAPI поддерживает асинхронность из коробки. Это позволяет обрабатывать множество запросов одновременно, не блокируя сервер.

В этой статье мы разберём:

Что такое асинхронность
async/await в FastAPI
Когда использовать async
Практические примеры

Содержание

Что нужно знать перед началом
Основная часть
Синхронный vs асинхронный код
Когда использовать async
Асинхронная работа с базой данных
Асинхронные HTTP-запросы
WebSockets
Несколько асинхронных задач
Асинхронный и синхронный код вместе
Полный пример
Задачи для закрепления
Нюансы и подводные камни
Частые ошибки и как их избежать
Ошибка 1: Использование time.sleep() в асинхронной функции
Ошибка 2: Забыл await перед вызовом асинхронной функции
Ошибка 3: Использование синхронной библиотеки в асинхронном эндпоинте
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого урока тебе понадобится:

Установленный FastAPI
Базовое понимание Python

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

Основная часть

Синхронный vs асинхронный код

Синхронный код (блокирующий):

import time

@app.get("/sync")
def sync_endpoint():
    time.sleep(2)  # Блокирует весь сервер!
    return {"message": "Готово"}

Асинхронный код (неблокирующий):

import asyncio

@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(2)  # Не блокирует сервер
    return {"message": "Готово"}

Совет: В асинхронном режиме сервер может обрабатывать другие запросы во время ожидания.

Когда использовать async

Используй async, когда:

Работаешь с базами данных (асинхронные драйверы)
Отправляешь HTTP-запросы (httpx, aiohttp)
Работаешь с WebSockets
Используешь внешние API

Используй синхронный код, когда:

Выполняешь простые вычисления
Работаешь с синхронными библиотеками
Читаешь локальные файлы

Асинхронная работа с базой данных

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

app = FastAPI()

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@app.get("/users")
async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(text("SELECT * FROM users"))
        users = result.fetchall()
        return {"users": users}

Асинхронные HTTP-запросы

import httpx

@app.get("/external")
async def get_external():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

WebSockets

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Сообщение: {data}")

Несколько асинхронных задач

import asyncio

async def fetch_data():
    await asyncio.sleep(2)
    return {"data": "данные"}

async def fetch_status():
    await asyncio.sleep(1)
    return {"status": "ok"}

@app.get("/parallel")
async def parallel_requests():
    # Запускаем задачи параллельно
    data_task = fetch_data()
    status_task = fetch_status()

    # Ожидаем завершения всех задач
    data, status = await asyncio.gather(data_task, status_task)

    return {"data": data, "status": status}

Асинхронный и синхронный код вместе

from fastapi import FastAPI
import time
import asyncio

app = FastAPI()

# Синхронная функция
def sync_task():
    return {"sync": "done"}

# Асинхронная функция
async def async_task():
    await asyncio.sleep(1)
    return {"async": "done"}

@app.get("/mixed")
async def mixed():
    # Синхронный код можно выполнять напрямую
    sync_result = sync_task()

    # Асинхронный код нужно await
    async_result = await async_task()

    return {"sync_result": sync_result, "async_result": async_result}

Полный пример

from fastapi import FastAPI
import asyncio
import httpx
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

app = FastAPI()

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# Синхронная функция
def cpu_task():
    result = 0
    for i in range(1000000):
        result += i
    return {"cpu_result": result}

# Асинхронная функция
async def external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

@app.get("/sync")
def sync_endpoint():
    return cpu_task()

@app.get("/async")
async def async_endpoint():
    data = await external_api()
    return {"external": data}

@app.get("/mixed")
async def mixed_endpoint():
    cpu_result = cpu_task()
    external_data = await external_api()

    async with AsyncSessionLocal() as session:
        result = await session.execute(text("SELECT * FROM users"))
        users = result.fetchall()

    return {
        "cpu": cpu_result,
        "external": external_data,
        "users": users
    }

@app.get("/parallel")
async def parallel_endpoint():
    task1 = external_api()
    task2 = external_api()
    task3 = external_api()

    results = await asyncio.gather(task1, task2, task3)

    return {"results": results}

Задачи для закрепления

Задача 1. Создай асинхронный эндпоинт с задержкой 1 секунда.

Задача 2. Создай эндпоинт, который выполняет два асинхронных запроса параллельно.

Задача 3. Напиши асинхронный эндпоинт для работы с внешним API.

Задача 4. Создай WebSocket эндпоинт.

Задача 5. В чём разница между синхронным и асинхронным кодом?

Ответы:

Задача 1.

import asyncio

@app.get("/delay")
async def delay():
    await asyncio.sleep(1)
    return {"message": "Готово"}

Задача 2.

@app.get("/parallel")
async def parallel():
    task1 = asyncio.sleep(1)
    task2 = asyncio.sleep(1)
    await asyncio.gather(task1, task2)
    return {"message": "Готово"}

Задача 3.

import httpx

@app.get("/external")
async def external():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

Задача 4.

@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Сообщение: {data}")

Задача 5.

# Синхронный код выполняется последовательно, блокируя сервер.
# Асинхронный код позволяет выполнять задачи параллельно, не блокируя сервер.

Нюансы и подводные камни

async def — асинхронная функция. def — синхронная.
В асинхронных функциях можно использовать await.
time.sleep() блокирует сервер. Используй asyncio.sleep().
Асинхронность полезна для I/O операций (БД, HTTP, WebSockets).
Для CPU-интенсивных задач асинхронность не даёт выигрыша.
Смешивать синхронный и асинхронный код можно, но с осторожностью.
В асинхронных эндпоинтах можно вызывать синхронные функции.

Частые ошибки и как их избежать

Ошибка 1: Использование `time.sleep()` в асинхронной функции

Неправильно:

@app.get("/")
async def endpoint():
    time.sleep(2)   # Блокирует сервер
    return {"ok": True}

Правильно:

await asyncio.sleep(2)

Ошибка 2: Забыл `await` перед вызовом асинхронной функции

result = await async_func()

Ошибка 3: Использование синхронной библиотеки в асинхронном эндпоинте

Для I/O операций используй асинхронные библиотеки (httpx, asyncpg, aiosqlite).

Шпаргалка

Что нужно	Как пишется
Асинхронная функция	`async def func():`
Ожидание	`await coroutine()`
Задержка	`await asyncio.sleep(seconds)`
Параллельные задачи	`asyncio.gather(task1, task2)`
Асинхронный HTTP	`httpx.AsyncClient()`
Асинхронная БД	`asyncpg`, `aiosqlite`