FastAPI

Дмитрий Шурмакин — ML Engineer

FastAPI - это легко, просто, лаконично!

FastAPI

– это современный веб-фреймворк для Python, который был разработан с целью обеспечения высокой производительности и эффективности в разработке веб-сервисов. Он основан на стандартах Python 3.6+ и полностью совместим с асинхронными возможностями языка.

Преимущества FastAPI:

  • Простота использования: FastAPI предоставляет простой и понятный способ создания API с помощью использования стандартных практик и синтаксиса Python, таких как аннотации типов данных и декораторы маршрутизации.
  • Производительность: FastAPI обеспечивает высокую производительность, в том числе, благодаря использованию асинхронных запросов.
  • Поддержка валидации данных: С использованием библиотеки Pydantic FastAPI автоматически выполняет валидацию входных данных и их сериализацию, что помогает предотвратить ошибки и упростить обработку запросов.
  • Интерактивная документация: Одним из преимуществ FastAPI является его способность автоматически генерировать интерактивную документацию API на основе стандарта OpenAPI (Swagger), что делает процесс разработки и тестирования API более удобным и прозрачным.

Недостатки FastAPI:

  • Относительная новизна: Поскольку FastAPI является сравнительно новым фреймворком, он имеeт меньшее количество сторонних библиотек и дополнений по сравнению с более устоявшимися аналогами, такими как Flask или Django. Но это нивелируется широким функционалом из коробки.

Пример приложения FastAPI:

Это простой пример использования FastAPI для создания минимальногр веб-приложения.

from fastapi import FastAPI

app = FastAPI()

@app.get("/", name="hello")
async def root():
    return {"message": "Hello World"}
  • from fastapi import FastAPI: импортируем класс FastAPI из библиотеки FastAPI, который используется для создания веб-приложений.
  • app = FastAPI(): Здесь создается экземпляр приложения FastAPI, который будет использоваться для определения маршрутов и обработки запросов.
  • @app.get("/", name="hello"): Это декоратор, который связывает функцию root() с маршрутом / и указывает на то, что она обрабатывает GET-запросы.
  • def root(): Это определение функции обработчика для маршрута /. Она возвращает JSON-ответ с текстом сообщения.

Запуск командой

### запуск приложения с помощью `fastapi-cli`
fastapi run main.py

### запуск приложения с помощью ASGI-сервера `uvicorn`
uvicorn main:app

Основные объекты и функции FastAPI подробно с примерами кода:

  • Маршруты (Routes)
  • APIRouter
  • Методы HTTP запросов (Methods)
  • Модели Pydantic
  • Ответы (Responses)
  • Статусы ответов (Response Statuses)
  • Фоновые задачи (Background Tasks)
  • Работа с файлами
  • События запуска и завершения (Startup/Shutdown Events)
  • Миддлвари (Middleware)

Маршруты (Routes)

Маршруты определяют, как приложение обрабатывает запросы. Они связывают функции обработчики с конкретными path’ами (сервиса). В FastAPI они определяются с помощью декораторов и объектами API.

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

В этом примере мы определяем маршрут GET /items/{item_id}, который принимает параметр item_id и возвращает JSON-объект с этим параметром.

APIRouter

FastAPI предоставляет удобный инструмент для структурирования вашего приложения, сохраняя при этом всю гибкость. Для использования APIRouter, вы создаете экземпляр класса APIRouter и определяете на нем маршруты. Затем вы можете подключить этот роутер к основному экземпляру FastAPI с помощью метода include_router.

Пример:

from fastapi import FastAPI, APIRouter

app = FastAPI()

v1_router = APIRouter(prefix="/v1")

@v1_router.get("/get_message/")
def get_message():
    """
    Some endpoint.
    """
    return {
        "message": "Hello from APIRouter",
    }

v2_router = APIRouter(prefix="/v2")

@v2_router.get("/get_message/")
def get_message_new():
    """
    Some endpoint.
    """
    return {
        "message": "Hello from APIRouter",
        "response_date": "12/04/2024",
    }

app.include_router(v1_router, prefix="/api")
app.include_router(v2_router, prefix="/api")

В этом примере:

  • Мы создаем экземпляр APIRouter с именем v1_router и v2_router.
  • Определяем маршрут GET / на v1_router, который возвращает сообщение “Hello from APIRouter”.
  • Определяем маршрут GET / на v2_router, который возвращает сообщение “Hello from APIRouter” дополнительным параметром обновленной версии API.
  • Подключаем v1_router к основному приложению FastAPI с префиксом /api/v1/get_message/.
  • Подключаем v2_router к основному приложению FastAPI с префиксом /api/v2/get_message/.

Теперь, когда вы запустите приложение FastAPI, вы сможете обратиться к маршруту GET /api/v2/get_message/ для получения ответа от APIRouter.

Использование APIRouter делает ваш код более организованным и управляемым, особенно когда ваше API растет в сложности и размере.

Методы HTTP запросов (Methods)

  • FastAPI поддерживает следующие методы HTTP: GET, POST, PUT, DELETE, PATCH, OPTIONS и HEAD.
  • Каждый метод имеет свое назначение и используется для выполнения определенных действий на сервере.
    • GET для получения данных,
    • POST для создания новых ресурсов,
    • PUT для обновления существующих ресурсов,
    • DELETE для удаления ресурсов,
    • PATCH для частичного обновления ресурсов,
    • OPTIONS для получения информации о возможностях сервера,
    • HEAD для получения заголовков без тела ответа.

Пример:

@app.post("/items/")
def create_item(item: Item):
    return item

Полный список методов можно найти в RFC 9110. Это документ, который описывает стандарты.

Соблюдение этого RFC не является обязательным, и методы могут быть использованы по своему усмотрению, например, можно обрабатывать создание ресурсов с помощью POST.

Важно помнить, что даже стандартные методы можно использовать гибко, а также при необходимости можно создать кастомные методы для специфичных задач.

Модели Pydantic

Модели Pydantic используются для определения структуры данных входящих и исходящих запросов. Они обеспечивают валидацию данных и автоматическую сериализацию между форматами JSON и Python.

Класс ConfigDict в Pydantic используется для определения дополнительных настроек модели данных. Атрибут json_schema_extra этого класса позволяет добавлять дополнительные метаданные или примеры данных для документации по API, улучшая читаемость и понятность документации.

Пример:

from pydantic import BaseModel, ConfigDict

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

    model_config = ConfigDict(
        json_schema_extra={
            "example": {
                "name": "товар #1",
                "description": "описание товара #1",
                "price": 17.50,
                "tax": 8.5,
            }
        }
    )

В этом примере мы создаем модель Item, которая содержит поля name, description, price и tax. Поля name и price обязательны для заполнения, а поля description и tax являются необязательными. Pydantic автоматически выполняет валидацию и преобразование типов данных, если это необходимо.

Передача параметров в FastAPI

FastAPI позволяет определять параметры входящих запросов. Вот несколько примеров:

  1. Параметры пути (Path parameters): Определяются в URL-адресе и используются для передачи данных.

    from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

    Отправка запроса

    import requests

url = "http://example.com/items/123"
response = requests.get(url)

    FastAPI самостоятельно парсит параметры полученные из запроса, сопоставляя типы на основе аннотаций.

  2. Параметры запроса (Query parameters): Передаются в URL-адресе через символ вопроса и амперсанд.

    from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

    Отправка запроса

    import requests

url = "http://example.com/items"
params = {"item_id": 123, "q": "search_query"}
response = requests.get(url, params=params)

# "http://example.com/items/?item_id=123&q=search_query"

  3. Тело запроса (Request Body): Передается в теле HTTP-запроса и используется для передачи структурированных данных, например, JSON или формы.

    from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return item

    Отправка запроса

    import requests

url = "http://example.com/items"
payload = {"name": "Example Item", "description": "Description of the item", "price": 10.99}
response = requests.post(url, json=payload)

Ответы (Responses)

FastAPI предоставляет удобные методы для создания ответов на запросы. Например, JSONResponse для возврата JSON-данных или HTMLResponse для возврата HTML-контента.

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

@app.get("/items/")
def read_item(item_id: int):
    return JSONResponse(
        status_code=200,
        content={"name": "Item 1", "item_id": item_id, "description": "Item description"}
    )

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None

@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user

В примере выше, метод create_user принимает данные о пользователе в виде модели UserIn и возвращает данные о созданном пользователе в виде модели UserOut. Использование response_model=UserOut указывает на то, что FastAPI должен проверить, что возвращаемые данные соответствуют модели UserOut и преобразует их в JSON согласно структуре модели.

Статусы ответов (Response Statuses)

FastAPI предоставляет простой доступ к кодам статуса ответов HTTP, таким как 200 (OK), 404 (Not Found) и т. д.

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    # Обработка создания элемента...
    return {"message": "Item created successfully"}

Здесь мы возвращаем статус создания ресурса (201 Created) вместе с сообщением об успешном создании элемента. Мы так же можем передавать целое число, которое соответствует коду ошибки.

Фоновые задачи (Background Tasks)

Фоновые задачи позволяют выполнять асинхронные операции в фоне во время обработки запросов.

Особенности и пример использования фоновых задач в FastAPI:

  1. Обработка в фоне:
    • Фоновые задачи выполняются асинхронно в фоновом режиме, не блокируя ответ на HTTP запрос и основной поток выполнения приложения.
    • Это позволяет приложению продолжать обслуживать запросы и взаимодействовать с клиентами, даже во время выполнения долгосрочных операций.
  2. Очередь задач:
    • FastAPI предоставляет возможность добавлять фоновые задачи в специальную очередь, которая автоматически выполняет их по мере возможности.
  3. Взаимодействие с запросами:
    • Фоновые задачи могут получать данные из HTTP запросов, например, параметры пути, параметры запроса или тело запроса, и использовать их для выполнения в фоновом режиме.
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def send_logs(data: dict):
    # отправка данных в БД...


@app.post("/data/")
async def create_data(data: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_logs, data)
    return {"message": "Data received and processing started"}

Этот пример демонстрирует добавление фоновой задачи для обработки полученных данных без блокировки основного потока выполнения.

Но есть несколько случаев, когда может быть разумным переходить от использования Background Tasks к Celery и т.п. инструментам:

  • Высокая нагрузка и требования к масштабируемости
  • Долгие и ресурсоемкие задачи
  • Отложенная обработка задач
  • Мониторинг и управление задачами
  • Повторная обработка и планирование задач

Работа с файлами

FastAPI предоставляет удобные методы для получения файлов из запросов и их обработки.

from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return {"filename": file.filename, "contents": contents}

Этот пример позволяет загружать файлы и возвращать их содержимое.

💡 необходимо установить модуль python-multipart который скачивает батчами файлы.

События запуска и завершения (Startup/Shutdown Events)

FastAPI предоставляет возможность определять обработчики событий при запуске и остановке приложения с помощью так называемых “lifespan events” (событий жизненного цикла). Эти события позволяют выполнять определенные действия при запуске и остановке сервера, что может быть полезно, например, для подготовки и очистки ресурсов.

Вот более подробное описание событий при запуске и остановке приложения в FastAPI:

  1. Startup Events (События при запуске):
    • Эти события происходят при запуске приложения перед тем, как начнется прослушивание запросов.
    • Обработчики событий при запуске могут выполнять различные действия, такие как установка соединения с базой данных, загрузка конфигураций, настройка сервисов и т. д.
    • Для определения обработчика события при запуске используется декоратор @app.on_event("startup").
  2. Shutdown Events (События при остановке):
    • Эти события происходят при остановке приложения после того, как прекращается прослушивание запросов.
    • Обработчики событий при остановке могут выполнять действия по очистке ресурсов, закрытию соединений, сохранению состояния и т. д.
    • Для определения обработчика события при остановке также используется декоратор @app.on_event("shutdown").
from fastapi import FastAPI
from loguru import logger

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    app.state.client_db = create_client()
    logger.info("Application initialized")

@app.on_event("shutdown")
async def shutdown_event():
        app.state.client_db.close()
    logger.info("Application shutdown")

В поздних версиях предлагается их заменить на функцию lifespan, в которой описывается логика операций при старте и завершении приложения.

async def lifespan(app: FastAPI):
    # Логика при запуске
    app.state.client_db = create_client()
    logger.info("Application initialized")
    yield
    # Логика при остановке
    app.state.client_db.close()
    logger.info("Application shutdown")

app = FastAPI(lifespan=lifespan)

Миддлвари (Middleware)

Middleware - представляет собой механизм для обработки запросов до и после их обработки соответствующим обработчиком маршрута. Это позволяет выполнять дополнительные действия, такие как аутентификация, логирование, обработка ошибок и многое другое, на каждом этапе обработки запроса.

Типы middleware:

  1. Промежуточное ПО до обработки запроса (Request Middleware):
    • Middleware, применяемое до обработки запроса, позволяет выполнить действия над запросом до передачи его соответствующему обработчику маршрута.
    • Это может быть использовано для аутентификации пользователей, проверки доступа, предварительной обработки данных запроса и т. д.
  2. Промежуточное ПО после обработки запроса (Response Middleware):
    • может быть использовано для логирования, обработки ошибок, изменения формата ответа и т. д.

Рассмотрим пример создания middleware в Starlette, который будет отправлять логи с мета-информацией о запросе в базу данных, вам нужно будет создать функцию, которая будет обернута в middleware. Вот пример кода, который может быть использован для этой цели:

from starlette.middleware.base import (
    BaseHTTPMiddleware,
    RequestResponseEndpoint,
)
from starlette.requests import Request
from starlette.responses import Response
import json
from datetime import datetime

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        # Формируем словарь с мета-информацией
        meta_info = {
            "timestamp": datetime.now(),
            "client_ip": request.client.host,
            "cookies": request.cookies,
            "user_agent": request.headers.get("user-agent"),
        }

        await self.save_to_db(meta_info)

        # Продолжаем обработку запроса
        response = await call_next(request)
        return response

    async def save_to_db(self, meta_info: str) -> None:
        """Функция для записи мета-информации в базу данных."""
        pass

# Затем добавьте middleware в FastAPI-приложение
from fastapi import FastAPI

app = FastAPI()

app.add_middleware(LoggingMiddleware)

В этом примере, LoggingMiddleware является классом, который наследуется от BaseHTTPMiddleware и определяет как middleware будет работать. Функция dispatch вызывается каждый раз, когда приходит запрос, и она может использоваться для записи мета-информации о запросе в базу данных.

Обратите внимание, что save_to_db - это функция, для работы с базами данных асинхронно, вы можете использовать библиотеки, такие как asyncpg для PostgreSQL или aiomysql для MySQL.