Розробка і тестування асинхронного API з FastAPI та Pytest

39 хв. читання

Про що цей посібник

Після прочитання матеріалу ви навчитесь:

  • розробці асинхронного 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 на момент написання матеріалу).

Важливі особливості фреймворку:

  1. Його розробники надихались Flask, тож FastAPI містить легкий мікрофреймворк з підтримкою декораторів, подібних на Flask-роути.
  2. Використовує переваги підказок типів у Python для оголошення параметрів, що дозволяє робити валідацію (через Pydantic) та створювати OpenAPI/Swagger-документацію.
  3. Створений на базі Starlette, підтримує розробку асинхронних API.
  4. Швидкий. Оскільки асинхронна модель набагато продуктивніша за традиційну синхронну, фреймворк може позмагатися з 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. Далі ми встановлюємо робочу директорію з двома змінними середовища:

  1. PYTHONDONTWRITEBYTECODE: попереджає Python від запису pyc-файлів до disc (еквівалент: опція python -B);
  2. 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 запуститься з такими налаштуваннями:

  1. --reload активує автооновлення, тобто сервер автоматично перезапускатиметься після змін в коді;
  2. --workers 1 забезпечує єдиний воркер-процес;
  3. --host 0.0.0.0 визначає адресу машини, яка хостить сервер;
  4. --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.

Розробка і тестування асинхронного API з FastAPI та Pytest

Налаштування тестового середовища

Створіть теку 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. Він підтримує такі методи:

  1. database.fetch_all(query)
  2. database.fetch_one(query)
  3. database.iterate(query)
  4. database.execute(query)
  5. 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 для створення нових записів, яка приймає об'єкт корисного навантаження, а потім:

  1. Створює об'єкт запиту insert SQLAlchemy ;
  2. Виконує запит та повертає згенерований 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 =========================================

Додаткова валідація

Додатково перевіримо роути на дотримання таких умов:

  1. Параметр id більший за 0 для перегляду нотатки, її оновлення та видалення;
  2. Поля 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:

  1. ... — означає, що значення обов'язкове (Ellipsis);
  2. gt — значення повинно бути більшим за 0.

Тепер тест має пройти успішно. Перевіримо також API-документацію:

Розробка і тестування асинхронного API з FastAPI та Pytest

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 значно полегшується.

Перевірте, наскільки ви тепер обізнані з темою, пройшовшись пунктами на початку матеріалу.

Шукаєте більше інформації?

  1. Огляньте офіційний посібник. Він досить об'ємний, проте вартий прочитання.
  2. Розширте функціонал застосунку, додавши асинхронне фонове завдання, міграції БД та автентифікацію.
  3. Винесіть конфігурацію застосунку в окремий файл.
  4. В продакшен-середовищі вам, мабуть, знадобиться налаштувати Gunicorn та передати йому управління Uvicorn. Огляньте матеріал про запуск застосунку з Gunicorn, а також посібник з деплою для заглиблення в тему. Додатково огляньте офіційний образ Docker uvicorn-gunicorn-fastapi.

Сирцевий код можна знайти в репозиторії fastapi-crud-async.

Помітили помилку? Повідомте автору, для цього достатньо виділити текст з помилкою та натиснути Ctrl+Enter
Codeguida 5.2K
Приєднався: 9 місяців тому
Коментарі (0)

    Ще немає коментарів

Щоб залишити коментар необхідно авторизуватися.

Вхід / Реєстрація