Привіт! Мене звуть Ярослав Мартиненко, я 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. Успіхів!
If you have found a spelling error, please, notify us by selecting that text and pressing Ctrl+Enter.
Favbet Tech – це ІТ-компанія зі 100% українською ДНК, що створює досконалі сервіси для iGaming і Betting з використанням передових технологій та надає доступ до них. Favbet Tech розробляє інноваційне програмне забезпечення через складну багатокомпонентну платформу, яка здатна витримувати величезні навантаження та створювати унікальний досвід для гравців.
Цей матеріал – не редакційний, це – особиста думка його автора. Редакція може не поділяти цю думку.
Сообщить об опечатке
Текст, который будет отправлен нашим редакторам: