asynchronous-programming/code/asyncio_fastapi_code_example.py - master - mksmakarov/learn-python

asynchronous-programming/code/asyncio_fastapi_code_example.py
69 строк3 KB
noir3cnt
example of asyncio with FastAPI
07 июн 2026, 15:41
07 июн 2026, 15:41626fc23
import asyncio
import time
from typing import List

# --- Импортируем необходимые модули ---
from fastapi import FastAPI
import httpx  # Асинхронная библиотека для HTTP-запросов

app = FastAPI()

# --- Константы ---
WEATHER_API_URL_TEMPLATE = "https://wttr.in/{city}?format=j1"
CITIES = ["Лондон", "Париж", "Токио", "Нью-Йорк"]

# --- Вспомогательная асинхронная функция ---
async def fetch_weather(client: httpx.AsyncClient, city: str) -> dict:
    """
    Получает данные о погоде для одного города.
    Принимает уже созданный клиент httpx, чтобы не создавать новое соединение.
    """
    url = WEATHER_API_URL_TEMPLATE.format(city=city)
    response = await client.get(url)
    response.raise_for_status()  # Вызовет исключение, если статус != 200
    return {"city": city, "temperature": response.json()["current_condition"][0]["temp_C"]}

# --- Синхронный эндпоинт (для сравнения) ---
@app.get("/sync-weather")
def get_sync_weather():
    """
    Этот эндпоинт использует стандартную библиотеку requests,
    которая является блокирующей. Он НЕ будет использовать преимущества FastAPI.
    """
    import requests # Импорт здесь для чистоты примера
    
    start_time = time.time()
    weather_data = []

    for city in CITIES:
        r = requests.get(WEATHER_API_URL_TEMPLATE.format(city=city))
        data = r.json()
        temp = data["current_condition"][0]["temp_C"]
        weather_data.append({"city": city, "temperature": temp})

    elapsed = time.time() - start_time
    print(f"--- Синхронный запрос обработан за {elapsed:.2f} секунд ---")
    
    return {"data": weather_data, "processed_in": f"{elapsed:.2f}s"}

# --- Асинхронный эндпоинт (рекомендуемый способ) ---
@app.get("/async-weather")
async def get_async_weather():
    """
    Этот эндпоинт использует асинхронность.
    Все запросы к API отправляются практически одновременно,
    а затем мы ждем их завершения.
    """
    start_time = time.time()

    async with httpx.AsyncClient() as client:
        # Создаем список корутин (будущих задач)
        tasks = [fetch_weather(client, city) for city in CITIES]
        
        # Запускаем все задачи конкурентно и ждем их результатов
        results = await asyncio.gather(*tasks)

    elapsed = time.time() - start_time
    print(f"\n+++ Асинхронный запрос обработан за {elapsed:.2f} секунд +++")

    return {"data": results, "processed_in": f"{elapsed:.2f}s"}