Что требуется, чтобы попытаться получить от этого материала какое-то подобие удовольствия

• вы знаете что такое virtual environment
• вы знаете как устанавливать зависимости, используя файл requirements.txt
• вы имели дело с Django или Flask (не обязательно, но это поможет)
• знаете что такое Postman, curl

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

Часть 1: Hello World

Делаем бесполезное приложение в образовательных целях

Давайте начнем с простого.

Нам нужно создать virtualenv.

pip install fastapi
pip install uvicorn

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

После этого создаем файл main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "hello world"}

Пока все кажется очень невинным 🙂 Но первый шаг сделан, еще немного — и впервые ступим на Луну.

А пока давайте запустим наш сервер и удостоверимся, что он работает.

uvicorn main:app --reload

В данной команде main — имя нашего файла с приложением, app — имя переменной, а --reload — аргумент, который запустит "наблюдающий" процесс. Если мы сделаем какие-то изменения в нашем коде и сохраним, сервер автоматически перезагрузится.

С помощью Postman или curl мы можем проверить работает ли наш эндпоинт.

Убиваем наш процесс, нажимая Ctrl + C . Пора двигаться дальше.

Добавляем dynamic routing в бесполезное приложение

Давайте попробуем теперь dynamic routing с аннотацией типов, который любезно предлагает нам фреймворк. Для этого я предлагаю сделать еще один раут (эндпоинт) в main.py.

...

@app.get("/dynamic_routing/{number}")
async def return_number(number: int): # type annotation syntax
    return {"number": number}

Тестируем.

Отлично. Если мы вводим параметр другого типа (string) наше приложение автоматически отрабатывает response и в довольно доступной форме поясняет клиенту (в данном случае Postman) в чем он был не прав.

Добавляем парсинг query parameters в наше бесполезное приложение

Модифицируем немного наш раут.

...

@app.get("/dynamic_routing/{number}")
async def return_number(number: int, add: int = 0, multiply: int = 1):
    return {"number": (number + add) * multiply}

Перестаем тестировать руками

Зачем тестировать вручную, если можно делать это с помощью кода?

Я предлагаю это делать с помощью pytest потому, что я заметил, что разработчики FastAPI рекомендуют его в своей документации. Pytest — удобная и интуитивно понятная вещь, поэтому почему бы и нет?

Устанавливаем в наш virtualenv новые зависимости.

pip install pytest
pip install requests

Создаем новый файл test_app.py — пока мы маленькое бесполезное приложение, одного файла будет более, чем достаточно.

from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "hello world"}

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

python3 -m pytest

Выглядит так, будто пока что все работает корректно. Давайте напишем тест на наш динамический раут и то, как он работает с query parameters.

def test_dynamic_routing():
    number: int = 10
    response = client.get(f"/dynamic_routing/{number}")
    assert response.status_code == 200
    assert response.json() == {"number": number}

def test_dynamic_routing_with_query_params():
    number: int = 10
    add: int = 12
    multiply: int = 4
    response = client.get(
        f"/dynamic_routing/{number}", params={"add": add, "multiply": multiply}
    )
    assert response.status_code == 200
    assert response.json() == {"number": (number + add) * multiply}

Запускаем новую тест сессию командой выше, и убеждаемся, что пока все работает 🙂 Наше бесполезное приложение все еще бесполезно, но приятно, что оно хотя бы ведет себя так, как мы ожидаем.

Добавляем POST route в наше бесполезное приложение

Давайте сделаем файл models.py, чтобы хранить наши модели отдельно от раутов. Наше приложение хоть и маленькое, но лично мне кажется, что так гораздо удобнее.

from pydantic import BaseModel

class SoftwareEngineer(BaseModel):
    main_language: str
    years_experience: float
    likes_coffee: bool

Будем создавать инженеров с помощью нашего бесполезного приложения 🙂

Добавляем раут в main.py

...
from .models import SoftwareEngineer
...
@app.post("/software_engineers/")
async def new_engineer(engineer: SoftwareEngineer):
    return engineer

И сразу же пишем тест в test_app.py

...
def test_create_engineer():
    main_language: str = "Python"
    years_experience: int = 2
    likes_coffee: bool = True

    payload = {
        "main_language": main_language,
        "years_experience": years_experience,
        "likes_coffee": likes_coffee,
    }

    response = client.post("/software_engineers/", json=payload)
    assert response.status_code == 200
    assert response.json() == payload

Запускаем тест сессию и — ура, все снова работает 🙂 Приятно же?

Расширяем модель

Давайте попробуем добавить пароли к нашим Software Engineer-ам 🙂 Посмотрим, что из этого получится, и как с этим быть.

В файле models.py

class SoftwareEngineer(BaseModel):
    main_language: str
    years_experience: float
    likes_coffee: bool
    password: str # new !

Расширяем наш тест в файле test_app.py.

...
def test_create_engineer():
    main_language: str = "Python"
    years_experience: int = 2
    likes_coffee: bool = True
    password: str = "mysupersecretpassword" # new!

    payload = {
        "main_language": main_language,
        "years_experience": years_experience,
        "likes_coffee": likes_coffee,
        "password": password, # new!
    }

    response = client.post("/software_engineers/", json=payload)
    assert response.status_code == 200
    assert response.json() == payload
    assert "password" in response.json().keys() # new!

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

Модифицируем наш раут в main.py

...
@app.post(
    "/software_engineers/",
    response_model=SoftwareEngineer, # new !
    response_model_exclude=["password"], # new !
)
async def new_engineer(engineer: SoftwareEngineer):
    return engineer

И проверяем есть ли пароль в респонсе с помощью наших тестов в test_app.py

...

def test_create_engineer():
    main_language: str = "Python"
    years_experience: int = 2
    likes_coffee: bool = True
    password: str = "mysupersecretpassword"

    payload = {
        "main_language": main_language,
        "years_experience": years_experience,
        "likes_coffee": likes_coffee,
        "password": password,
    }

    response = client.post("/software_engineers/", json=payload)
    assert response.status_code == 200
    # assert response.json() == payload
    assert "password" not in response.json().keys() # new !

Все работает 🙂

Часть 2. Подключаем Tortoise ORM к нашему FastAPI приложению

SqlAlchemy — это восхитительная, мощная ORM. Ее рекомендуют разработчики FastAPI и вы наверняка уже успели ее воспользоваться и знаете ее синтаксис.

Но я подумал, мы же тут вроде как учимся и делаем бесполезные приложения ради приключений, разве нет? Почему бы во имя духа приключений не попробовать бы подключить асинхронную ORM Tortoise и к ней туда же — Aerich для миграций.

Tortoise ORM — начальный уровень

Я честно скажу, что это было непросто, но, используя навыки сыщика, я нашел, как легко и без боли интегрировать Tortoise ORM в FastAPI приложение

pip install tortoise-orm
pip install argon2 # для хеширования паролей

В файле main.py

...
from tortoise.contrib.fastapi import register_tortoise
...
# наше приложение
# и рауты
# здесь
# и в самом низу
# регистрируем Sqlite базу данных

register_tortoise(
    app,
    db_url="sqlite://:memory:",
    modules={"models": ["tortoise_models"]},
    generate_schemas=True,
    add_exception_handlers=True,
)

Создаем файл tortoise_models.py

from tortoise.models import Model
from tortoise import fields
from tortoise.contrib.pydantic import pydantic_model_creator
from pydantic import BaseModel

class SoftwareEngineers(Model):
    uuid = fields.UUIDField(pk=True)
    loves_coffee = fields.BooleanField(default=True)
    years_experience = fields.FloatField()
    main_language = fields.CharField(max_length=128)
    password_hash = fields.CharField(max_length=500)
    created_at = fields.DatetimeField(auto_now_add=True)
    modified_at = fields.DatetimeField(auto_now_add=True)

SoftwareEngineer_Pydantic = pydantic_model_creator(
    SoftwareEngineers, name="SoftwareEngineer"
)

Тут мы определяем нашу ORM модель. Для более удобной сериализации входящего request body я добавил Pydantic модель внизу (если знаете более красивый способ сделать это — сообщите мне).

Все еще файл tortoise_models.py

...
class SoftwareEngineerIn(BaseModel):
    loves_coffee: bool
    years_experience: float
    password: str
    main_language: str

Меняем наш раут в main.py

@app.post(
    "/software_engineers/",
    response_model=SoftwareEngineer_Pydantic,
    response_model_exclude=["password_hash"],
)
async def new_engineer(engineer: SoftwareEngineerIn):
    password_hash = argon2.argon2_hash(engineer.password, "some_salt")
    engineer_dict = engineer.dict()
    engineer_dict.update({"password_hash": password_hash})
    engineer_dict.pop("password")
    software_eng_obj = await SoftwareEngineers.create(**engineer_dict)
    return await SoftwareEngineer_Pydantic.from_tortoise_orm(software_eng_obj)

Я думаю, это не самая элегантная имплементация, но это должно сработать. Я добавил хеширование пароля с помощью алгоритма argon2.

Прочитал недавно о том, что argon2 стал алгоритмом-победителем крипто-соревнования. У нас же тут приключения, новые фреймворки и новые ORM библиотеки, поэтому мне показалось, что argon2 отлично впишется в компанию.

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

Пишем асинхронные тесты для нашей асинхронной ORM

Настало время для асинхронных тестов.

pip install asynctest

Файл test_app.py

import asyncio
from tortoise_models import SoftwareEngineers
from typing import Generator

import pytest
from fastapi.testclient import TestClient
from main import app

from tortoise.contrib.test import finalizer, initializer

@pytest.fixture(scope="module")
def client() -> Generator:
    initializer(["tortoise_models"])
    with TestClient(app) as c:
        yield c
    finalizer()

@pytest.fixture(scope="module")
def event_loop(client: TestClient) -> Generator:
    yield client.task.get_loop()

...
# все те тесты, что у нас были, остаются, кроме последнего
...

def test_create_engineer(client: TestClient, event_loop: asyncio.AbstractEventLoop):
    main_language: str = "Python"
    years_experience: int = 2
    likes_coffee: bool = True
    password: str = "mysupersecretpassword"

    payload = {
        "main_language": main_language,
        "years_experience": years_experience,
        "loves_coffee": likes_coffee,
        "password": password,
    }

    response = client.post("/software_engineers/", json=payload)
    data = response.json()
    assert response.status_code == 200
    assert "password_hash" not in data.keys()

    async def get_engineer_by_db():
        engineer = await SoftwareEngineers.get(uuid=data.get("uuid"))
        return engineer

    eng_obj = event_loop.run_until_complete(get_engineer_by_db())

    assert str(eng_obj.uuid) == data.get("uuid")

Запускаем тест-сессию. Все тесты должны реализоваться успешно 🙂

Заключение обзора

Если вы нашли ошибки в тексте или в коде, вы можете сделать PR в открытом репозитории этого блога, или написать мне на почту — kohler.messages[собака]gmail.com

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

...то подписывайтесь на мою рассылку!

Не забывайте читать документации зависимостей, которые вы себе добавляете в проект 🙂