Про що цей посібник
Після прочитання матеріалу ви навчитесь:
- розробці асинхронного RESTful API з Python та FastAPI;
- розробці на основі тестів (TDD – Test-Driven Development);
- тестуванню застосунку на FastAPI з Pytest;
- асинхронній взаємодії з базою даних Postgres;
- контейнеризації FastAPI та Postgres за допомогою Docker;
- параметризації тестових функцій та моканню функціоналу в тестах з Pytest;
- правильному документуванню RESTful API за допомогою Swagger/OpenAPI.
FastAPI
FastAPI — це сучасний та продуктивний фреймворк Python, що ідеально підходить для створення RESTful API. Він може обробляти як синхронні, так і асинхронні запити, а ще містить вбудовану підтримку валідації даних, серіалізації JSON, автентифікації та авторизації, а також OpenAPI-документації (версія 3.0.2 на момент написання матеріалу).
Важливі особливості фреймворку:
- Його розробники надихались Flask, тож FastAPI містить легкий мікрофреймворк з підтримкою декораторів, подібних на Flask-роути.
- Використовує переваги підказок типів у Python для оголошення параметрів, що дозволяє робити валідацію (через Pydantic) та створювати OpenAPI/Swagger-документацію.
- Створений на базі Starlette, підтримує розробку асинхронних API.
- Швидкий. Оскільки асинхронна модель набагато продуктивніша за традиційну синхронну, фреймворк може позмагатися з Node та Go в продуктивності.
Огляньте перелік фіч фреймворку FastAPI з офіційної документації для глибшого розуміння. Корисним буде також переглянути «Альтернативи, Натхнення та Порівняння» для зіставлення FastAPI з іншими вебфреймворками та технологіями.
Налаштування проєкту
Почнемо зі створення теки проєкту. Назвемо її fastapi-crud. Додамо файл docker-compose.yml та теку src до кореня проєкту. В src також додамо Dockerfile, requirements.txt та теку app. Наостанок розташуємо в app файли __init__.py та main.py.
Остаточна структура щойно створеного проєкту буде такою:
fastapi-crud
├── docker-compose.yml
└── src
├── Dockerfile
├── app
│ ├── __init__.py
│ └── main.py
└── requirements.txt
На відміну від Django чи Flask, у FastAPI немає вбудованого сервера розробки. Тож ми використовуватимемо Uvicorn, сервер ASGI для запуску FastAPI.
Новачок в ASGI? Ознайомтеся з чудовим вступом до нього.
Додамо версії FastAPI та Uvicorn до файлу з вимогами:
fastapi==0.46.0
uvicorn==0.11.1
На думку автора, той факт, що FastAPI не підтримує з коробки сервер розробки, досить неоднозначний. З одного боку, необхідно більше зусиль, аби запустити застосунок в режимі розробки. З іншого — маємо концептуальне розділення вебфреймворка та вебсервера, що часто плутає новачків, які переходять з фреймворків, котрі мають вбудований сервер розробки.
Перейдемо до main.py, де створюємо екземпляр FastAPI та тестовий роут:
from fastapi import FastAPI
app = FastAPI()
@app.get("/ping")
def pong():
return {"ping": "pong!"}
Встановлюємо Docker, якщо досі його немає на машині. Пропишемо налаштування в Dockerfile:
# підтягуємо офіційний базовий образ
FROM python:3.8.1-alpine
# встановлюємо робочу директорію
WORKDIR /usr/src/app
# встановлюємо змінні середовища
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
# копіюємо файл з вимогами
COPY ./requirements.txt /usr/src/app/requirements.txt
# встановлюємо залежності
RUN set -eux \\
&& apk add --no-cache --virtual .build-deps build-base \\
libressl-dev libffi-dev gcc musl-dev python3-dev \\
&& pip install --upgrade pip setuptools wheel \\
&& pip install -r /usr/src/app/requirements.txt \\
&& rm -rf /root/.cache/pip
# копіюємо проєкт
COPY . /usr/src/app/
Тож ми почали з Docker-образу на базі Alpine для Python 3.8.1. Далі ми встановлюємо робочу директорію з двома змінними середовища:
-
PYTHONDONTWRITEBYTECODE: попереджає Python від запису pyc-файлів до disc (еквівалент: опціяpython -B); -
PYTHONUNBUFFERED: попереджає Python від буферизації потоку стандартного виводу (stdout) та стандартного потоку помилок (еквівалентно до опціїpython -u).
Наприкінці ми скопіювали файл requirements.txt, встановили деякі залежності системного рівня, оновили pip, встановили залежності самого застосунку та скопіювали безпосередньо FastAPI-застосунок.
Подивіться документацію Docker для Python-розробників для глибшого розуміння структури Docker-файлів, а також для найкращих конфігурацій саме для розробки на Python.
Наступним кроком додаємо файл docker-compose.yml до кореня проєкту.
version: '3.7'
services:
web:
build: ./src
command: uvicorn app.main:app --reload --workers 1 --host 0.0.0.0 --port 8000
volumes:
- ./src/:/usr/src/app/
ports:
- 8002:8000
Коли ми підіймемо контейнер, Uvicorn запуститься з такими налаштуваннями:
-
--reloadактивує автооновлення, тобто сервер автоматично перезапускатиметься після змін в коді; -
--workers 1забезпечує єдиний воркер-процес; -
--host 0.0.0.0визначає адресу машини, яка хостить сервер; -
--port 8000визначає порт для хостингу сервера.
Рядок app.main:app вказує Uvicorn де знайти застосунок FastAPI ASGI, тобто «всередині модуля app», app = FastAPI(), у файлі main.py.
Більше про налаштування Docker Compose за посиланням.
Приводимо збірку образу та підіймаємо контейнер:
$ docker-compose up -d --build
Перейдіть за посиланням http://localhost:8002/ping, аби побачити:
{
"ping": "pong!"
}
Ви також можете переглянути інтерактивну документацію API, розроблену Swagger UI, перейшовши на http://localhost:8002/docs.
Налаштування тестового середовища
Створіть теку tests в src та додайте туди файли __init__.py та test_main.py.
from starlette.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_ping():
response = client.get("/ping")
assert response.status_code == 200
assert response.json() == {"ping": "pong!"}
У фрагменті вище ми імпортували TestClient від Starlette, який використовує бібліотеку Requests для створення запитів у FastAPI-застосунку.
Одразу додаємо Pytest та Requests до requirements.txt.
fastapi==0.46.0
uvicorn==0.11.1
# dev
pytest==5.3.2
requests==2.22.0
Оновлюємо образ, а потім запускаємо тести:
$ docker-compose up -d --build
$ docker-compose exec web pytest .
Як результат ви повинні побачити:
======================================== test session starts ========================================
platform linux -- Python 3.8.1, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: /usr/src/app
collected 1 item
tests/test_main.py . [100%]
========================================= 1 passed in 0.15s =========================================
Перш ніж підемо далі, додамо Pytest-фікстуру до нового файлу src/tests/conftest.py:
import pytest
from starlette.testclient import TestClient
from app.main import app
@pytest.fixture(scope="module")
def test_app():
client = TestClient(app)
yield client # тестування тут
Оновимо файли тестів, щоб вони використовували щойно створену фікстуру:
def test_ping(test_app):
response = test_app.get("/ping")
assert response.status_code == 200
assert response.json() == {"ping": "pong!"}
Структура нашого проєкту тепер така:
fastapi-crud
├── docker-compose.yml
└── src
├── Dockerfile
├── app
│ ├── __init__.py
│ └── main.py
├── requirements.txt
└── tests
├── __init__.py
├── conftest.py
└── test_main.py
Асинхронні обробники
Перейдемо до перетворення синхронного обробника на асинхронний.
Замість того щоб створювати собі проблему і запускати чергу завдань (на зразок Celery or RQ) чи використовувати потоки, FastAPI підтримує асинхронні роути. Якщо у вас немає блокувальних дій вводу/виводу в обробнику, ви можете оголосити його асинхронним з ключовим словом async:
@app.get("/ping")
async def pong():
# тут можуть бути деякі асинхронні операції
# приклад: `notes = await get_all_notes()`
return {"ping": "pong!"}
На цьому все. Оновіть код вашого обробника та переконайтеся, що тести досі проходять.
======================================== test session starts ========================================
platform linux -- Python 3.8.1, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: /usr/src/app
collected 1 item
tests/test_main.py . [100%]
========================================= 1 passed in 0.14s =========================================
Якщо хочете заглибитись в технічну сторону питання асинхронності, то посібник з Конкурентності та async/await до ваших послуг.
Роути
Далі створимо базові CRUD-роути, враховуючи найкращі практики RESTful API.
| Ендпоінт | HTTP-метод | CRUD-метод | Результат |
|---|---|---|---|
/notes/ |
GET | READ | отримання всіх записів |
/notes/:id/ |
GET | READ | отримання певного запису |
/notes/ |
POST | CREATE | створення запису |
/notes/:id/ |
PUT | PUT | оновлення запису |
/notes/:id/ |
DELETE | DELETE | видалення запису |
Для кожного з роутів, ми:
- напишемо тест;
- запустимо тест, щоб переконатися, що він провалився;
- напишемо ще трохи коду, аби тест пройшов;
- відрефакторимо код (за необхідності).
Перед тим як зануритись в роботу, організуємо певну структуру, аби краще спроєктувати CRUD-роути з APIRouter від FastAPI.
Ви можете як розділити на модулі великі проєкти, так і додати версіонування до вашого API з APIRouter. Якщо ви знайомі з Flask, то це еквівалент Blueprint.
Спершу додамо нову теку з назвою api, яка буде дочірньою для app. Створимо там вже славнозвісний __init__.py.
Тепер ми можемо перемістити роут /ping до файлу src/app/api/ping.py:
from fastapi import APIRouter
router = APIRouter()
@router.get("/ping")
async def pong():
# тут можуть бути деякі асинхронні операції
# приклад: `notes = await get_all_notes()`
return {"ping": "pong!"}
Далі внесемо зміни до main.py, аби замінити старий роут та поєднати роутер з нашим основним застосунком.
from fastapi import FastAPI
from app.api import ping
app = FastAPI()
app.include_router(ping.router)
Змінимо назву test_main.py на test_ping.py.
Переконайтесь, що адреси http://localhost:8002/ping та http://localhost:8002/docs досі працюють. Не забудьте також перевірити, чи проходять тести, перш ніж йти далі.
fastapi-crud
├── docker-compose.yml
└── src
├── Dockerfile
├── app
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ └── ping.py
│ └── main.py
├── requirements.txt
└── tests
├── __init__.py
├── conftest.py
└── test_ping.py
Налаштування Postgres
Для конфігурації Postgres нам необхідно додати новий сервіс в docker-compose.yml та відповідні змінні середовища, а також встановити asyncpg.
Спершу додамо новий сервіс під назвою db в docker-compose.yml :
version: '3.7'
services:
web:
build: ./src
command: uvicorn app.main:app --reload --workers 1 --host 0.0.0.0 --port 8000
volumes:
- ./src/:/usr/src/app/
ports:
- 8002:8000
environment:
- DATABASE_URL=postgresql://hello_fastapi:hello_fastapi@db/hello_fastapi_dev
db:
image: postgres:12.1-alpine
volumes:
- postgres_data:/var/lib/postgresql/data/
environment:
- POSTGRES_USER=hello_fastapi
- POSTGRES_PASSWORD=hello_fastapi
- POSTGRES_DB=hello_fastapi_dev
volumes:
postgres_data:
Щоб зберігати дані поза життєвим циклом контейнера, ми сконфігуруємо том. Таке налаштування прив'яже postgres_data до директорії /var/lib/postgresql/data/ в контейнері.
Ми також додали ключ середовища, аби визначити стандартну назву бази даних та встановити ім'я користувача та пароль.
Для більш детальної інформації зверніться до секції «Змінні середовища» за посиланням.
Вказуємо в Dockerfile встановити пакети, необхідні для asyncpg:
# підтягуємо офіційний базовий образ
FROM python:3.8.1-alpine
# встановлюємо робочу директорію
WORKDIR /usr/src/app
# встановлюємо змінні середовища
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
# копіюємо файл з вимогами
COPY ./requirements.txt /usr/src/app/requirements.txt
# встановлюємо залежності
RUN set -eux \\
&& apk add --no-cache --virtual .build-deps build-base \\
libressl-dev libffi-dev gcc musl-dev python3-dev \\
postgresql-dev \\
&& pip install --upgrade pip setuptools wheel \\
&& pip install -r /usr/src/app/requirements.txt \\
&& rm -rf /root/.cache/pip
# копіюємо проєкт
COPY . /usr/src/app/
Додаємо asyncpg до src/requirements.txt:
asyncpg==0.20.0
fastapi==0.46.0
uvicorn==0.11.1
# dev
pytest==5.3.2
requests==2.22.0
Оглянемо вміст файлу db.py в src/app:
import os
from databases import Database
from sqlalchemy import create_engine, MetaData
DATABASE_URL = os.getenv("DATABASE_URL")
# SQLAlchemy
engine = create_engine(DATABASE_URL)
metadata = MetaData()
# конструктор запитів бази даних
database = Database(DATABASE_URL)
Використовуючи URI бази даних та щойно сконфігуровані в Docker Compose файлі облікові дані, ми створили рушій SQLAlchemy (потрібен для спілкування з базою даних) поруч з екземпляром Metadata (потрібен для створення схеми БД). Ми також створили новий екземпляр Database з конструктора databases.
databases — це асинхронний конструктор SQL-запитів, який працює поверх мови виразів SQLAlchemy Core. Він підтримує такі методи:
-
database.fetch_all(query) -
database.fetch_one(query) -
database.iterate(query) -
database.execute(query) -
database.execute_many(query)
Огляньте посібник з асинхронних (реляційних) баз даних SQL та документацію бази даних Starlette для ознайомлення з деталями асинхронної роботи з базами даних.
Як завжди, оновлюємо файл з вимогами:
asyncpg==0.20.0
databases[postgresql]==0.2.6
fastapi==0.46.0
SQLAlchemy==1.3.12
uvicorn==0.11.1
# dev
pytest==5.3.2
requests==2.22.0
Моделі
Модель SQLAlchemy
Додамо модель app до src/app/db.py:
import os
from sqlalchemy import (Column, DateTime, Integer, MetaData, String, Table,
create_engine)
from sqlalchemy.sql import func
from databases import Database
DATABASE_URL = os.getenv("DATABASE_URL")
# SQLAlchemy
engine = create_engine(DATABASE_URL)
metadata = MetaData()
notes = Table(
"notes",
metadata,
Column("id", Integer, primary_key=True),
Column("title", String(50)),
Column("description", String(50)),
Column("created_date", DateTime, default=func.now(), nullable=False),
)
# конструктор запитів бази даних
database = Database(DATABASE_URL)
Прив'яжемо модель до бази даних у файлі main.py, а також додамо обробники підключення та роз'єднання з БД.
from fastapi import FastAPI
from app.api import ping
from app.db import engine, metadata, database
metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
app.include_router(ping.router)
Створюємо новий образ та запускаємо два контейнери:
$ docker-compose up -d --build
Переконаймося, що таблиця notes була створена:
$ docker-compose exec db psql --username=hello_fastapi --dbname=hello_fastapi_dev
psql (12.1)
Type "help" for help.
hello_fastapi_dev=# \\l
List of databases
Name | Owner | Encoding | Collate | Ctype | Access privileges
-------------------+---------------+----------+------------+------------+---------------------------------
hello_fastapi_dev | hello_fastapi | UTF8 | en_US.utf8 | en_US.utf8 |
postgres | hello_fastapi | UTF8 | en_US.utf8 | en_US.utf8 |
template0 | hello_fastapi | UTF8 | en_US.utf8 | en_US.utf8 | =c/hello_fastapi +
| | | | | hello_fastapi=CTc/hello_fastapi
template1 | hello_fastapi | UTF8 | en_US.utf8 | en_US.utf8 | =c/hello_fastapi +
| | | | | hello_fastapi=CTc/hello_fastapi
(4 rows)
hello_fastapi_dev=# \\c hello_fastapi_dev
You are now connected to database "hello_fastapi_dev" as user "hello_fastapi".
hello_fastapi_dev=# \\dt
List of relations
Schema | Name | Type | Owner
--------+-------+-------+---------------
public | notes | table | hello_fastapi
(1 row)
hello_fastapi_dev=# \\q
Модель Pydantic
Вперше використовуєте Pydantic? Ознайомтесь з посібником з офіційної документації.
Створимо Pydantic модель NoteSchema з двома обов'язковими полями, title та description. Розмістимо їх у фалі models.py у теці src/app/api:
from pydantic import BaseModel
class NoteSchema(BaseModel):
title: str
description: str
NoteSchema використовуватиметься для валідації корисного навантаження для створення та оновлення записів.
POST-роут
Зійдемо трохи зі звичного для TDD-розробки процесу для першого роуту, щоб зрозуміти шаблон, який використовуватимемо для останніх роутів.
Код
Створимо новий файл notes.py в директорії src/app/api:
from app.api import crud
from app.api.models import NoteDB, NoteSchema
from fastapi import APIRouter, HTTPException
router = APIRouter()
@router.post("/", response_model=NoteDB, status_code=201)
async def create_note(payload: NoteSchema):
note_id = await crud.post(payload)
response_object = {
"id": note_id,
"title": payload.title,
"description": payload.description,
}
return response_object
Тут ми оголосили обробник, який очікує корисне навантаження у форматі payload: NoteSchema із заголовком та описанням.
Коли на роут направляється POST-запит, FastAPI зчитує тіло запиту та валідує дані:
- якщо дані валідні, вони будуть доступними в параметрі
payload. FastAPI також генерує визначення JSON-схеми, які потім потрібні для автоматичної генерації OpenAPI схеми та API-документації; - якщо дані не пройшли валідацію, миттєво повертається помилка.
Для більшої інформації, огляньте документацію про тіло запиту.
Варто помітити, що ми використовували ключове слово async, враховуючи асинхронний зв'язок з БД. Іншими словами, у нас немає блокувальних операцій вводу/виводу в обробнику.
Далі створимо новий файл з назвою crud.py у тій самій теці.
from app.api.models import NoteSchema
from app.db import notes, database
async def post(payload: NoteSchema):
query = notes.insert().values(title=payload.title, description=payload.description)
return await database.execute(query=query)
Ми додали функцію post для створення нових записів, яка приймає об'єкт корисного навантаження, а потім:
- Створює об'єкт запиту
insertSQLAlchemy ; - Виконує запит та повертає згенерований ID.
Далі нам необхідно визначити нову модель Pydantic для використання її як response_model.
@router.post("/", response_model=NoteDB, status_code=201)
Оновимо файл з моделями:
from pydantic import BaseModel
class NoteSchema(BaseModel):
title: str
description: str
class NoteDB(NoteSchema):
id: int
Модель NoteDB наслідується від раніше створеної NoteSchema і додає поле id.
Об'єднаємо новий роутер в main.py:
from app.api import notes, ping
from app.db import database, engine, metadata
from fastapi import FastAPI
metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
app.include_router(ping.router)
app.include_router(notes.router, prefix="/notes", tags=["notes"])
Зверніть увагу на префікс URL, а також тег "notes", який додасться до схеми OpenAPI (для операцій групування).
Протестуємо з curl або HTTPie:
$ http --json POST http://localhost:8002/notes/ title=foo description=bar
Ви повинні побачити:
HTTP/1.1 201 Created
content-length: 42
content-type: application/json
date: Thu, 09 Jan 2020 19:03:48 GMT
server: uvicorn
{
"description": "bar",
"id": 1,
"title": "foo"
}
Ви також можете тестувати створені ендпоінти вручну за адресою http://localhost:8002/docs.
Тест
Напишемо такий тест у файлі src/tests/test_notes.py:
import json
import pytest
from app.api import crud
def test_create_note(test_app, monkeypatch):
test_request_payload = {"title": "something", "description": "something else"}
test_response_payload = {"id": 1, "title": "something", "description": "something else"}
async def mock_post(payload):
return 1
monkeypatch.setattr(crud, "post", mock_post)
response = test_app.post("/notes/", data=json.dumps(test_request_payload),)
assert response.status_code == 201
assert response.json() == test_response_payload
def test_create_note_invalid_json(test_app):
response = test_app.post("/notes/", data=json.dumps({"title": "something"}))
assert response.status_code == 422
Тут ми використали фікстуру monkeypatch, щоб замокати функцію crud.post. Далі ми припускаємо, що ендпоінт відповідає очікуваним статус-кодом та тілом запиту.
$ docker-compose exec web pytest .
======================================== test session starts ========================================
platform linux -- Python 3.8.1, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: /usr/src/app
collected 3 items
tests/test_notes.py .. [ 66%]
tests/test_ping.py . [100%]
========================================= 3 passed in 0.26s =========================================
Тепер ми можемо сконфігурувати всі інші CRUD-роути, використовуючи підхід TDD:
fastapi-crud
├── docker-compose.yml
└── src
├── Dockerfile
├── app
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── crud.py
│ │ ├── models.py
│ │ ├── notes.py
│ │ └── ping.py
│ ├── db.py
│ └── main.py
├── requirements.txt
└── tests
├── __init__.py
├── conftest.py
├── test_notes.py
└── test_ping.py
GET-роути
Тест
Додамо тести:
def test_read_note(test_app, monkeypatch):
test_data = {"id": 1, "title": "something", "description": "something else"}
async def mock_get(id):
return test_data
monkeypatch.setattr(crud, "get", mock_get)
response = test_app.get("/notes/1")
assert response.status_code == 200
assert response.json() == test_data
def test_read_note_incorrect_id(test_app, monkeypatch):
async def mock_get(id):
return None
monkeypatch.setattr(crud, "get", mock_get)
response = test_app.get("/notes/999")
assert response.status_code == 404
assert response.json()["detail"] == "Note not found"
Звичайно, поки що вони проваляться:
======================================== test session starts ========================================
platform linux -- Python 3.8.1, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: /usr/src/app
collected 5 items
tests/test_notes.py ..FF [ 80%]
tests/test_ping.py . [100%]
============================================= FAILURES ==============================================
__________________________________________ test_read_note ___________________________________________
test_app = <starlette.testclient.TestClient object at 0x7f60d5ca1910>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f60d5ca19a0>
def test_read_note(test_app, monkeypatch):
test_data = {"id": 1, "title": "something", "description": "something else"}
async def mock_get(id):
return test_data
> monkeypatch.setattr(crud, "get", mock_get)
E AttributeError: <module 'app.api.crud' from '/usr/src/app/app/api/crud.py'> has no attribute 'get'
tests/test_notes.py:34: AttributeError
____________________________________ test_read_note_incorrect_id ____________________________________
test_app = <starlette.testclient.TestClient object at 0x7f60d5ca1910>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f60d5c9c7c0>
def test_read_note_incorrect_id(test_app, monkeypatch):
async def mock_get(id):
return None
> monkeypatch.setattr(crud, "get", mock_get)
E AttributeError: <module 'app.api.crud' from '/usr/src/app/app/api/crud.py'> has no attribute 'get'
tests/test_notes.py:45: AttributeError
==================================== 2 failed, 3 passed in 0.36s ====================================
Код
Додамо обробник:
@router.get("/{id}/", response_model=NoteDB)
async def read_note(id: int):
note = await crud.get(id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
return note
Замість того щоб передавати корисне навантаження, ми тут передаємо id цілочислового типу, яке береться зі шляху: наприклад /notes/5/.
Додамо функцію, яка оброблятиме get-запит, до файлу crud.py:
async def get(id: int):
query = notes.select().where(id == notes.c.id)
return await database.fetch_one(query=query)
Перш ніж рухатись далі, переконаймося, що тести проходять та вручну протестуємо новий ендпоінт у браузері з curl чи HTTPie (можна також використати API-документацію).
Тест
Не забуваємо додати тест, аби перевірити функціонал повернення всіх нотаток:
def test_read_all_notes(test_app, monkeypatch):
test_data = [
{"title": "something", "description": "something else", "id": 1},
{"title": "someone", "description": "someone else", "id": 2},
]
async def mock_get_all():
return test_data
monkeypatch.setattr(crud, "get_all", mock_get_all)
response = test_app.get("/notes/")
assert response.status_code == 200
assert response.json() == test_data
Знову ж таки, тест спершу провалюється.
Код
@router.get("/", response_model=List[NoteDB])
async def read_all_notes():
return await crud.get_all()
Імпортуємо List з модуля typing у Python:
from typing import List
response_model – це наш List з підтипом NoteDB.
Додамо CRUD-функцію:
async def get_all():
query = notes.select()
return await database.fetch_all(query=query)
Переконаймося, що автоматичний тест пройдений. Не забувайте також перевіряти ендпоінти вручну.
PUT -роут
Тест
def test_update_note(test_app, monkeypatch):
test_update_data = {"title": "someone", "description": "someone else", "id": 1}
async def mock_get(id):
return True
monkeypatch.setattr(crud, "get", mock_get)
async def mock_put(id, payload):
return 1
monkeypatch.setattr(crud, "put", mock_put)
response = test_app.put("/notes/1/", data=json.dumps(test_update_data))
assert response.status_code == 200
assert response.json() == test_update_data
@pytest.mark.parametrize(
"id, payload, status_code",
[
[1, {}, 422],
[1, {"description": "bar"}, 422],
[999, {"title": "foo", "description": "bar"}, 404],
],
)
def test_update_note_invalid(test_app, monkeypatch, id, payload, status_code):
async def mock_get(id):
return None
monkeypatch.setattr(crud, "get", mock_get)
response = test_app.put(f"/notes/{id}/", data=json.dumps(payload),)
assert response.status_code == status_code
В наведених тестах ми використали декоратор parametrize від Pytest, аби параметризувати аргументи для функції test_update_note_invalid.
Код
Напишемо код, який пройде тест:
@router.put("/{id}/", response_model=NoteDB)
async def update_note(id: int, payload: NoteSchema):
note = await crud.get(id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
note_id = await crud.put(id, payload)
response_object = {
"id": note_id,
"title": payload.title,
"description": payload.description,
}
return response_object
Допоміжна функція:
async def put(id: int, payload: NoteSchema):
query = (
notes
.update()
.where(id == notes.c.id)
.values(title=payload.title, description=payload.description)
.returning(notes.c.id)
)
return await database.execute(query=query)
DELETE-роут
Тест
Як зазичай, спочатку пишемо тест:
def test_remove_note(test_app, monkeypatch):
test_data = {"title": "something", "description": "something else", "id": 1}
async def mock_get(id):
return test_data
monkeypatch.setattr(crud, "get", mock_get)
async def mock_delete(id):
return id
monkeypatch.setattr(crud, "delete", mock_delete)
response = test_app.delete("/notes/1/")
assert response.status_code == 200
assert response.json() == test_data
def test_remove_note_incorrect_id(test_app, monkeypatch):
async def mock_get(id):
return None
monkeypatch.setattr(crud, "get", mock_get)
response = test_app.delete("/notes/999/")
assert response.status_code == 404
assert response.json()["detail"] == "Note not found"
Код
Обробник:
@router.delete("/{id}/", response_model=NoteDB)
async def delete_note(id: int):
note = await crud.get(id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
await crud.delete(id)
return note
CRUD-утиліта:
async def delete(id: int):
query = notes.delete().where(id == notes.c.id)
return await database.execute(query=query)
І перевіряємо, чи пройдено тест:
======================================== test session starts ========================================
platform linux -- Python 3.8.1, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: /usr/src/app
collected 12 items
tests/test_notes.py ........... [ 91%]
tests/test_ping.py . [100%]
======================================== 12 passed in 0.56s =========================================
Додаткова валідація
Додатково перевіримо роути на дотримання таких умов:
- Параметр
idбільший за 0 для перегляду нотатки, її оновлення та видалення; - Поля
titleтаdescriptionз корисного навантаження запиту повинні бути довшими за 3 та коротшими за 50 символів при додаванні та оновленні нотатки.
GET
Оновимо тест test_read_note_incorrect_id:
def test_read_note_incorrect_id(test_app, monkeypatch):
async def mock_get(id):
return None
monkeypatch.setattr(crud, "get", mock_get)
response = test_app.get("/notes/999")
assert response.status_code == 404
assert response.json()["detail"] == "Note not found"
response = test_app.get("/notes/0")
assert response.status_code == 422
Тест повинен провалитись:
> assert response.status_code == 422
E assert 404 == 422
E + where 404 = <Response [404]>.status_code
Оновимо обробник:
@router.get("/{id}/", response_model=NoteDB)
async def read_note(id: int = Path(..., gt=0),):
note = await crud.get(id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
return note
Переконаймося, що імпортували Path:
from fastapi import APIRouter, HTTPException, Path
Тож ми додали такі метадані параметру з Path:
-
...— означає, що значення обов'язкове (Ellipsis); -
gt— значення повинно бути більшим за 0.
Тепер тест має пройти успішно. Перевіримо також API-документацію:
POST
Оновимо тест test_create_note_invalid_json:
def test_create_note_invalid_json(test_app):
response = test_app.post("/notes/", data=json.dumps({"title": "something"}))
assert response.status_code == 422
response = test_app.post("/notes/", data=json.dumps({"title": "1", "description": "2"}))
assert response.status_code == 422
Щоб тест пройшов, оновимо модель NoteSchema в такий спосіб:
class NoteSchema(BaseModel):
title: str = Field(..., min_length=3, max_length=50)
description: str = Field(..., min_length=3, max_length=50)
Тут ми додали додаткову валідацію для моделі Pydantic за допомогою Field. Все працює за аналогією з Path.
Імпортуємо потрібний модуль:
from pydantic import BaseModel, Field
PUT
Додамо більше сценаріїв, які необхідно перевірити, до тесту test_update_note_invalid:
@pytest.mark.parametrize(
"id, payload, status_code",
[
[1, {}, 422],
[1, {"description": "bar"}, 422],
[999, {"title": "foo", "description": "bar"}, 404],
[1, {"title": "1", "description": "bar"}, 422],
[1, {"title": "foo", "description": "1"}, 422],
[0, {"title": "foo", "description": "bar"}, 422],
],
)
def test_update_note_invalid(test_app, monkeypatch, id, payload, status_code):
async def mock_get(id):
return None
monkeypatch.setattr(crud, "get", mock_get)
response = test_app.put(f"/notes/{id}/", data=json.dumps(payload),)
assert response.status_code == status_code
Обробник:
@router.put("/{id}/", response_model=NoteDB)
async def update_note(payload: NoteSchema, id: int = Path(..., gt=0),):
note = await crud.get(id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
note_id = await crud.put(id, payload)
response_object = {
"id": note_id,
"title": payload.title,
"description": payload.description,
}
return response_object
DELETE
Тест:
def test_remove_note_incorrect_id(test_app, monkeypatch):
async def mock_get(id):
return None
monkeypatch.setattr(crud, "get", mock_get)
response = test_app.delete("/notes/999/")
assert response.status_code == 404
assert response.json()["detail"] == "Note not found"
response = test_app.delete("/notes/0/")
assert response.status_code == 422
Обробник:
@router.delete("/{id}/", response_model=NoteDB)
async def delete_note(id: int = Path(..., gt=0)):
note = await crud.get(id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
await crud.delete(id)
return note
Переконуємось, що тест пройшов:
======================================== test session starts ========================================
platform linux -- Python 3.8.1, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: /usr/src/app
collected 15 items
tests/test_notes.py .............. [ 93%]
tests/test_ping.py . [100%]
======================================== 15 passed in 0.45s =========================================
Синхронний приклад
Автор матеріалу створив також синхронну версію API для порівняння двох моделей. Код можна знайти в репозиторії за посиланням. Спробуйте також провести тестування продуктивності для обох версій самостійно за допомогою ApacheBench.
Висновок
У матеріалі ми розглянули процес розробки та тестування асинхронного API з FastAPI, Postgres, Pytest та Docker, використовуючи TDD-підхід.
FastAPI — це потужний фреймворк, який отримав простоту Flask, подібні до Django «батарейки» та продуктивність Go/Node. Завдяки цьому створення RESTful API значно полегшується.
Перевірте, наскільки ви тепер обізнані з темою, пройшовшись пунктами на початку матеріалу.
Шукаєте більше інформації?
- Огляньте офіційний посібник. Він досить об'ємний, проте вартий прочитання.
- Розширте функціонал застосунку, додавши асинхронне фонове завдання, міграції БД та автентифікацію.
- Винесіть конфігурацію застосунку в окремий файл.
- В продакшен-середовищі вам, мабуть, знадобиться налаштувати Gunicorn та передати йому управління Uvicorn. Огляньте матеріал про запуск застосунку з Gunicorn, а також посібник з деплою для заглиблення в тему. Додатково огляньте офіційний образ Docker uvicorn-gunicorn-fastapi.
Сирцевий код можна знайти в репозиторії fastapi-crud-async.
Ще немає коментарів