Чому я обираю FastAPI: основні можливості та переваги фреймворку

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

Рік тому я дізнався про FastAPI. Він є «спадкоємцем» філософії Flask, але вже «з коробки» надає цікаві фічі, про які я розповім у цій статті.

FastAPI не пропонує більше, ніж необхідний мінімум, тому розробник вільно може використовувати разом з цим фреймворком будь-які інструменти.

Що ж це за FastAPI

FastAPI — це відносно новий асинхронний вебфреймворк для Python. По суті, це гібрид Starlett та Pydantic.

Starlett — асинхронний вебфреймворк, Pydantic — бібліотека для валідації даних, серіалізації тощо. У документації FastAPI написано, що він може наблизитися за швидкістю до Node.js та Golang. Я цього не перевіряв, тому й вірити в це не буду. Для мене він швидкий з іншої причини. FastAPI дозволяє просто та оперативно написати невеликий REST AРІ, не витративши на це багато зусиль.

Давайте поглянемо, як легко (це лише моя суб’єктивна думка) можна почати роботу з 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
• 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 типу dictі та надали йому 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 у app.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. Успіхів!