Последнее обновление: 29 July, 2021
Если вам будет так комфортнее, вы можете посмотреть пошаговую реализацию проекта на гитхабе.
Давайте начнем с простого.
Нам нужно создать 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 с аннотацией типов, который любезно предлагает нам фреймворк. Для этого я предлагаю сделать еще один раут (эндпоинт) в main.py
.
...
@app.get("/dynamic_routing/{number}")
async def return_number(number: int): # type annotation syntax
return {"number": number}
Тестируем.
Отлично. Если мы вводим параметр другого типа (string) наше приложение автоматически отрабатывает response и в довольно доступной форме поясняет клиенту (в данном случае Postman) в чем он был не прав.
Модифицируем немного наш раут.
...
@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}
Запускаем новую тест сессию командой выше, и убеждаемся, что пока все работает 🙂 Наше бесполезное приложение все еще бесполезно, но приятно, что оно хотя бы ведет себя так, как мы ожидаем.
Давайте сделаем файл 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 !
Все работает 🙂
SqlAlchemy — это восхитительная, мощная ORM. Ее рекомендуют разработчики FastAPI и вы наверняка уже успели ее воспользоваться и знаете ее синтаксис.
Но я подумал, мы же тут вроде как учимся и делаем бесполезные приложения ради приключений, разве нет? Почему бы во имя духа приключений не попробовать бы подключить асинхронную ORM Tortoise и к ней туда же — Aerich для миграций.
Я честно скажу, что это было непросто, но, используя навыки сыщика, я нашел, как легко и без боли интегрировать 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 отлично впишется в компанию.
Вы можете заменить его любым другим алгоритмом хэширования паролей по своему вкусу и не устанавливать, как зависимость.
Настало время для асинхронных тестов.
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
Если вы хотите получить уведомление о том, когда я опубликую продолжение о том, как сделать что-то полезное на основе этого и разместить его в облаке...
...то подписывайтесь на мою рассылку!
Не забывайте читать документации зависимостей, которые вы себе добавляете в проект 🙂
Трой Кёлер - программист, живущий в Берлине, Германия. У него более 6 лет опыта работы в IT. Ранее он работал в одном из крупнейших интернет-магазинов Украины, а сейчас работает в Zalando. Он специализируется на языке программирования Rust, сложных бекенд системах, разработке продуктов и инженерных платформах.