Почему я выбираю FastAPI: основные возможности и преимущества фреймворка

Привет! Меня зовут Ярослав Мартыненко, я Python Developer в NIX. Раньше я занимался Embedded-разработкой, позже пошел в сторону веба. Уже больше года разрабатываю бэкенд на Python. Стараюсь постоянно изучать что-то новое и создавать то, что упростит жизнь окружающим.

Год назад я узнал о FastAPI. Он «наследник» философии Flask, но уже «из коробки» предоставляет интересные фичи, о которых я расскажу в этой статье.

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

Что же это за FastAPI

FastAPI — это относительно новый асинхронный веб-фреймворк для Python. По сути это гибрид Starlett и Pydantic.

Starlett — асинхронный веб-фреймворк, Pydantic – библиотека для валидации данных, сериализации и т.д. В документации FastAPI написано, что он может приблизиться по скорости к Node.js и Golang. Я этого не проверял, потому и верить в это не буду. Для меня он быстр по другой причине. FastAPI позволяет просто и оперативно написать небольшой REST API, не затратив на это много усилий.

Давайте посмотрим, как легко (это только мое субъективное мнение) можно начать работу с FastAPI.

Начало работы

В первую очередь стоит установить нужные нам зависимости, а это сам фреймворк и ASGI-сервер, поскольку у FastAPI нет встроенного сервера, как у Flask или Django. В документации предлагается использовать uvicorn в качестве ASGI-сервера:

pip install fastapi
pip install uvicorn

В FastAPI используется подобная Flask система объявления эндпоинтов — с помощью декораторов. Поэтому работающим с Flask будет достаточно легко приспособиться к FastAPI. Теперь создадим объект нашей программы и добавим роут HelloWorld:

from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
    return {"message": "Hello World"}

Мы объявили, что при GET-запросе на / мы вернем json {"message": "Hello World"} — особенных отличий от Flask здесь нет.

Важная ремарка: эндпоинт также можно объявить в синхронном стиле, используя просто def, если вы хотите использовать await. FastAPI разрулит все за вас. За что я люблю FastAPI — так это за его лаконичность.

Давайте объявим роут, который будет ожидать какой-либо параметр как часть пути:

@app.get("/item/{id}")
async def get_item(id):
    return id

Теперь, если мы перейдем по адресу /item/2, то получим 2 в ответ. А что делать, если кто-то захочет нам прислать вместо цифры, например, dva? Хотелось бы защитить себя от таких конфузов. И здесь нам приходит на помощь Python 3.6+ і type_hints.

Тайп-хинтинг (объявление типов) в целом помогает сделать код более понятным и позволяет использовать инструменты для статического анализа (такие, как mypy). FastAPI заставляет вас использовать тайп-хинтинг, тем самым улучшая качество кода и уменьшая вероятность того, что вы где-то по невнимательности допустили ошибку.

Теперь определим, что наш id должен быть типа  int:

@app.get("/item/{id}")
async def get_item(id: int):
    return id

Мы достаточно просто добавили валидацию и теперь можно попытаться передать dva и посмотреть, что же получится. В ответ получим сообщение, что сделали что-нибудь не так.

Сервер вернет нам 422 статус-код и следующий json:

{
    "detail": [
        {
            "loc": [
                "path",{
    "detail": [
        {
            "loc": [
                "path",
                "item_id"
            ],
            "msg": "value is not a valid integer",
            "type": "type_error.integer"
        }
    ]
}
                "item_id"
            ],
            "msg": "value is not a valid integer",
            "type": "type_error.integer"
        }
    ]
}

На этом этапе пришло время Pydantic. Он сгенерирует данные о том, где обнаружена ошибка, и подскажет, что мы сделали не так. Опять же, не всем придется по душе статус-код 422 и данные об ошибке, которые нам генерирует Pydantic. Но все это можно кастомизировать, если очень хочется.

А как объявить, что мы хотим какой-то квери-параметр, да еще чтобы он был необязательным? Все просто: если аргумент функции не объявлен как часть пути, FastAPI будет считать, что он должен быть получен как квери-параметр. Для того чтобы сделать его необязательным, придадим ему дефолтное значение.

Еще одна прекрасная фича FastAPI — то, что мы можем объявить, например, enum, чтобы задать определенные значения, которые ожидаем на вход:

class Framework(str, Enum):
    flask = "flask"
    django = "django"
    fastapi = "fastapi
@app.get("/framework")
def framework(framework: Framework = Framework.flask):
    return {"framework": framework}

Следующая интересная фича — преобразование типов. Если мы хотим получить булевое значение как квери-параметр, нам все равно придется его передать как число или строку. Рydantic предлагает превратить логически правильное значение в булевый тип вот так:

@app.get("/items")
async def read_item(short: bool = False):
    if short:
        return "Short items description"
    else:
        return "Full items description"

Для эндпоинта, указанного выше, следующие значения будут валидны и превращены в булевое значение True:

• http://127.0.0.1:8000/items?short=1
• http://127.0.0.1:8000/items?short=True
• http://127.0.0.1:8000/items?short=true
Освітній курс від laba: Delivery Manager.
Ведіть проекти до успішної доставки.
Дізнатись про курс
• http://127.0.0.1:8000/items?short=on
• http://127.0.0.1:8000/items?short=yes

Иногда нам нужно более гибко настраивать момент, где искать и откуда доставать параметры. Например, мы хотим извлечь значение из хедера. Для этого FastAPI предоставляет нам следующие инструменты: Query, Body, Path, Header, Cookie, импортируемые из FastAPI. Они помогают не только явно определить, где искать параметр, но и объявить дополнительную валидацию.

Давайте рассмотрим это на примере:

from typing import Optional
from fastapi import FastAPI, Query, Header
app = FastAPI()
@app.get("/")
async def test(number: Optional[int] = Query(None, alias="num", gt=0, le=10), owner: str = Header(...)):
    return {"number": number, "owner": owner}

Мы определили эндпоинт, ожидающий, что мы передадим ему число от 0 до 10 включительно как квери-параметр. Причем квери-параметр мы должны передавать как /?num=3, поскольку определили alias для этого параметра и теперь ожидаем, что он придет под именем num, и что у нас будет хедер Owner.

Pydantic-модели

Чаще всего, когда мы строим REST API, то хотим передавать какие-либо более сложные структуры в виде json в теле запроса. Эти структуры можно описать с помощью Рydantic-моделей.

Например, мы хотим принимать объект item, у которого есть имя, цена и опциональное описание:

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float

Также мы хотим добавить эндпоинт, который будет принимать POST-запросы, десериализировать и валидировать json и где-то хранить его. Наша модель Item — это класс, поэтому мы можем наследоваться от нее и создать модель, которая также будет содержать id. Ведь нам хочется сохранить где-то наш item. Ему присваивается id и уже вместе с этим мы можем вернуть клиенту ответ с кодом 201. 

Для начала создадим модель с новым полем id:

class ItemOut(Item):
    id: int

Далее — эндпоинт с аргументом item типа Item. Поскольку Item — это Pydantic-модель, FastAPI предполагает, что нам нужно достать item из тела запроса і content-type = application/json. Pydantic десериализирует эти данные и провалидирует их. Затем создадим объект типа ItemOut, у которого будет поле id, и вернем все это пользователю:

@app.post("/item/", response_model=ItemOut, status_code=201)
async def create_item(item: Item):
    item_with_id = ItemOut(**item.dict(), id=1)
    return item_with_id

Как вы можете увидеть в декораторе, мы определили, что возвращаемые данные будут типа ItemOut, а статус код —  201. Указание response_model необходимо для того, чтобы правильно сгенерировать документацию (об этом расскажу далее), а также сериализировать и провалидировать данные. Мы могли бы передать словарь вместо объекта ItemOut. Тогда FastAPI попытался превратить этот словарь в ItemOut-объект и провалидировать данные.

Если хотим создать более сложные структуры с вложенностью, то здесь тоже не возникает особого труда. Мы просто определяем нашу модель Pydantic, содержащую объекты с типом другой Pydantic–модели:

class OrderOut(BaseModel):
    id: int
    items: list[Item]

Еще одно преимущество FastAPI — автогенерация OpenApi-документации. Ничего не нужно подключать, не нужно танцевать с бубном — просто бери и пользуйся. По умолчанию документация находится по пути /docs.

Отложенные задачи

Иногда бывает, что мы хотим быстро вернуть респонс клиенту, а затратные задачи выполнить потом на фоне. Обычно для этого используется что-то вроде Celery или RQ. Чтобы не возиться с очередями и воркерами, у FastAPI есть такая фича, как background tasks. Мы можем объявить, что наша функция приобретает аргумент типа BackgroundTasks, и этот объект будет интерфейсом для создания бэкграунд-тасков:

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

На примере выше показана функция, которая что-то записывает в файл. Нам нужно обработать ее после того, как вернем пользователю респонс. Для этого объявим аргумент background_tasks с типом BackgroundTasksи с помощью него сможем прибавлять функции, которые нам нужно выполнить после того, как отработает наша view.

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

Инъекция зависимостей

FastAPI предоставляет систему для инъекции зависимостей в наши view. Для этого есть Depends. Зависимостью может быть callable–объект, в котором будет реализована определенная логика. Инжектируемый объект будет иметь доступ к контексту реквеста. Это означает, что мы сможем извлечь определенную общую логику из наших view и переиспользовать ее. 

Предлагаю рассмотреть этот процесс на примере:

from typing import Optional
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

Мы создали функцию, которая получает нам параметр для фильтрации и возвращает его как словарь. Затем подключили эту функцию как зависимость в наши view–функции read_itemsи read_users. Объявили аргумент типа commonи предоставили ему Depends (common_parameters).

Depends принимает как аргумент callable-объект, который вызовется перед обработкой нашего view. В этом случае он вернет словарь с параметрами фильтрации. Интересно здесь то, что нам безразлично, синхронная ли функция. Мы можем объявить common_parameters как синхронную и асинхронную. FastAPI все разрулит за нас.

Поскольку зависимостями могут быть callable-объекты, мы можем заменить нашу функцию, которая возвращает словарь с параметрами на что-то более элегантное:

class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit
@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    return commons

Как видите, мы заменили функцию на класс и теперь передаем его в Depends. В результате нам возвращается объект класса CommonQueryParams. Теперь мы можем получить доступ к его атрибутам через точку, например, commons.q. Вот так выглядят наши зависимости: 

По сути, это граф. Мы можем сделать его более сложным, где в текущие зависимости добавим другие и сделаем более специфическими. Предположим, у нас будет зависимость, которая проверяет, авторизован ли пользователь, и достает его из базы. Другая — связанная с первой зависимостью — проверяет, активен ли пользователь, а третья — которая зависит от второй — определяет, является ли он админом:

Поскольку это граф, возникает вопрос ромбовидной зависимости: сколько раз будет выполняться первая родительская зависимость? Ответ прост — всего раз. Обычно нам нужно только один раз выполнить действие над реквестом, а затем закэшировать данные, которые вернут зависимость. Это можно переопределить, передав в Dependsuse_cache=False:

def dep_a():
    logger.warning("A")
def dep_b(a = Depends(dep_a)):
    logger.warning("B")
def dep_c(a = Depends(dep_a, use_cache=False)):
    logger.warning("C")
@app.get("/test")
def test(dep_a = Depends(dep_a), dep_b = Depends(dep_b), dep_c = Depends(dep_c)):
    return "Hello world"

Зависимость dep_a идет первой в аргументах и ​​не имеет других зависимостей, поэтому она выполнится и кэширует ее. Зависимость dep_b идет следующей и имеет зависимость от dep_a, но вызов dep_aбыл сделан и ответ закэшен, поэтому dep_a не будет вызываться.

Далее следует dep_c, которая зависит от dep_a и определяет  use_cache=False для зависимости dep_a. Несмотря на то, что dep_a была закэшена, она все равно будет вызываться, и ответ также закэшивается. Затем вызовется dep_c. И только в конце выполнится наша функция test. 

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

async def get_db():
    logger.warning("Open connection")
    yield "database"
    logger.warning("Close connection")
async def task(database):
    logger.warning("Some task")
    logger.warning(f"DB: {database}")
@app.get("/test")
async def test(background_tasks: BackgroundTasks, database = Depends(get_db)):
    background_tasks.add_task(task, database)
    return database

Dependency Injector необходим для того, чтобы легко подменить нашу зависимость на mock. Предположим, эта зависимость — и есть клиент, который обращается к стороннему API по http. Делается это просто: подменяем возвращающую клиента зависимость на зависимость, которая возвращает mock с таким же публичным API.

Если у нас есть сервис для отправки сообщений, то при попытке запустить тесты с этим сервисом они упадут с ошибкой. Но мы можем определить pytest-фикстуру, в которой наша зависимость будет подменяться. Как это сделать? Добавим функцию, которая вернет mockв dependency_overrides, и после того, как тест сработает, очистим наши переопределения зависимостей app.dependency_overrides = {}:

import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def send_msg():
    raise ValueError("Error")
@app.get("/api")
def some_api(msg = Depends(send_msg)):
    return msg
@pytest.fixture
def mock_dependencies():
    def get_msg_mocked():
        return "Test pass"
    app.dependency_overrides[send_msg] = get_msg_mocked
    yield
    app.dependency_overrides = {}
@pytest.mark.usefixtures("mock_dependencies")
def test_my_api():
    res = client.get("/api")
    assert res.status_code == 200

Вывод

Я попытался кратко описать основные возможности FastAPI и показать, чем мне нравится этот фреймворк. Попробуйте его хотя бы для небольшого pet-проекта. Вокруг FastAPI достаточно быстро разрастается сообщество его поклонников, чуть ли не каждый день появляются новые библиотеки. Поэтому некоторые проекты постепенно переходят с Flask на FastAPI. Удачи!