Перейти до змісту

Best practices

Приклад з моїм проектом: https://github.com/samuel-edmund-morgan/py-fastapi-city-temperature-management-api

Найкращі практики FastAPI

Суб'єктивний список найкращих практик та конвенцій, які я використовую в стартапах.

Протягом останніх кількох років у продакшені ми приймали вдалі та невдалі рішення, які кардинально вплинули на наш досвід розробки. Деякими з них варто поділитися.

Зміст

Структура проекту

Є багато способів структурувати проект, але найкраща структура — це та, що є послідовною, зрозумілою та без сюрпризів.

Багато прикладів проектів та туторіалів ділять проект за типом файлу (наприклад, crud, routers, models), що добре працює для мікросервісів або проектів з меншою кількістю областей. Однак цей підхід не підходив для нашого моноліту з багатьма доменами та модулями.

Структура, яку я вважаю більш масштабованою та гнучкою для таких випадків, натхненна проектом Netflix Dispatch з незначними модифікаціями.

fastapi-project
├── alembic/
├── src
│   ├── auth
│   │   ├── router.py
│   │   ├── schemas.py  # pydantic моделі
│   │   ├── models.py  # моделі бд
│   │   ├── dependencies.py
│   │   ├── config.py  # локальні конфігурації
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── aws
│   │   ├── client.py  # модель клієнта для комунікації з зовнішніми сервісами
│   │   ├── schemas.py
│   │   ├── config.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   └── utils.py
│   └── posts
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   ├── dependencies.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── config.py  # глобальні конфігурації
│   ├── models.py  # глобальні моделі
│   ├── exceptions.py  # глобальні виключення
│   ├── pagination.py  # глобальний модуль, напр. пагінація
│   ├── database.py  # все, що стосується підключення до бд
│   └── main.py
├── tests/
│   ├── auth
│   ├── aws
│   └── posts
├── templates/
│   └── index.html
├── requirements
│   ├── base.txt
│   ├── dev.txt
│   └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini
  1. Зберігайте всі доменні директорії всередині папки src
  2. src/ — найвищий рівень додатку, містить спільні моделі, конфігурації, константи тощо.
  3. src/main.py — корінь проекту, який ініціалізує FastAPI додаток
  4. Кожен пакет має свій router, schemas, models тощо.
  5. router.py — ядро кожного модуля з усіма ендпоінтами
  6. schemas.py — для pydantic моделей
  7. models.py — для моделей бд
  8. service.py — бізнес-логіка, специфічна для модуля
  9. dependencies.py — залежності маршрутизатора
  10. constants.py — константи та коди помилок, специфічні для модуля
  11. config.py — напр. змінні середовища
  12. utils.py — функції, не пов'язані з бізнес-логікою, напр. нормалізація відповідей, збагачення даних тощо.
  13. exceptions.py — виключення, специфічні для модуля, напр. PostNotFound, InvalidUserData
  14. Коли пакету потрібні сервіси, залежності чи константи з інших пакетів — імпортуйте їх з явним ім'ям модуля
from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode  # якщо у нас є стандартний ErrorCode в модулі constants кожного пакету

Асинхронні маршрути

FastAPI — це насамперед асинхронний фреймворк. Він розроблений для роботи з асинхронними операціями вводу/виводу, і саме тому він такий швидкий.

Однак FastAPI не обмежує вас лише async маршрутами, і розробник може використовувати й sync маршрути. Це може збити з пантелику початківців, які вважають, що вони однакові, але це не так.

Задачі з інтенсивним вводом/виводом

Під капотом FastAPI може ефективно обробляти як async, так і sync операції вводу/виводу.

  • FastAPI запускає sync маршрути у пулі потоків, і блокуючі операції вводу/виводу не зупиняють цикл подій від виконання завдань.
  • Якщо маршрут визначений як async, він викликається звичайно через await, і FastAPI довіряє вам виконувати лише неблокуючі операції вводу/виводу.

Застереження полягає в тому, що якщо ви порушите цю довіру і виконаєте блокуючі операції в async маршрутах, цикл подій не зможе виконувати наступні завдання, поки ця блокуюча операція не завершиться.

import asyncio
import time

from fastapi import APIRouter


router = APIRouter()


@router.get("/terrible-ping")
async def terrible_ping():
    time.sleep(10) # Блокуюча операція вводу/виводу на 10 секунд, весь процес буде заблоковано

    return {"pong": True}

@router.get("/good-ping")
def good_ping():
    time.sleep(10) # Блокуюча операція вводу/виводу на 10 секунд, але в окремому потоці для всього маршруту `good_ping`

    return {"pong": True}

@router.get("/perfect-ping")
async def perfect_ping():
    await asyncio.sleep(10) # неблокуюча операція вводу/виводу

    return {"pong": True}

Що відбувається, коли ми викликаємо:

  1. GET /terrible-ping
  2. Сервер FastAPI отримує запит і починає його обробку
  3. Цикл подій сервера і всі завдання в черзі будуть чекати, поки time.sleep() завершиться
    1. Сервер вважає, що time.sleep() — це не операція вводу/виводу, тому він чекає її завершення
    2. Сервер не прийматиме нових запитів під час очікування
  4. Сервер повертає відповідь.
    1. Після відповіді сервер починає приймати нові запити
  5. GET /good-ping
  6. Сервер FastAPI отримує запит і починає його обробку
  7. FastAPI відправляє весь маршрут good_ping у пул потоків, де робочий потік виконає функцію
  8. Поки good_ping виконується, цикл подій вибирає наступні завдання з черги і працює з ними (напр. приймає новий запит, викликає бд)
    • Незалежно від головного потоку (тобто нашого додатку FastAPI), робочий потік буде чекати завершення time.sleep.
    • Синхронна операція блокує лише побічний потік, а не головний.
  9. Коли good_ping завершує роботу, сервер повертає відповідь клієнту
  10. GET /perfect-ping
  11. Сервер FastAPI отримує запит і починає його обробку
  12. FastAPI очікує asyncio.sleep(10)
  13. Цикл подій вибирає наступні завдання з черги і працює з ними (напр. приймає новий запит, викликає бд)
  14. Коли asyncio.sleep(10) завершено, сервер завершує виконання маршруту і повертає відповідь клієнту

[!WARNING] Примітки щодо пулу потоків:

  • Потоки потребують більше ресурсів, ніж корутини, тому вони не такі дешеві, як async операції вводу/виводу.
  • Пул потоків має обмежену кількість потоків, тобто ви можете вичерпати потоки і ваш додаток стане повільним. Детальніше (зовнішнє посилання)

Задачі з інтенсивним використанням CPU

Друге застереження полягає в тому, що операції, які є неблокуючими awaitables або надсилаються в пул потоків, повинні бути задачами вводу/виводу (напр. відкриття файлу, виклик бд, виклик зовнішнього API).

  • Очікування CPU-інтенсивних задач (напр. складні обчислення, обробка даних, транскодування відео) не має сенсу, оскільки CPU повинен працювати для завершення задач, тоді як операції вводу/виводу є зовнішніми і сервер нічого не робить, чекаючи їх завершення, тому він може перейти до наступних задач.
  • Запуск CPU-інтенсивних задач в інших потоках також неефективний через GIL. Коротко, GIL дозволяє працювати лише одному потоку одночасно, що робить його марним для CPU-задач.
  • Якщо ви хочете оптимізувати CPU-інтенсивні задачі, вам слід відправляти їх робочим процесам в іншому процесі.

Пов'язані питання на StackOverflow від спантеличених користувачів

  1. https://stackoverflow.com/questions/62976648/architecture-flask-vs-fastapi/70309597#70309597
  2. Тут також можна перевірити мою відповідь
  3. https://stackoverflow.com/questions/65342833/fastapi-uploadfile-is-slow-compared-to-flask
  4. https://stackoverflow.com/questions/71516140/fastapi-runs-api-calls-in-serial-instead-of-parallel-fashion

Pydantic

Активно використовуйте Pydantic

Pydantic має багатий набір можливостей для валідації та перетворення даних.

На додаток до звичайних можливостей, таких як обов'язкові та необов'язкові поля зі значеннями за замовчуванням, Pydantic має вбудовані інструменти обробки даних, такі як regex, enum, маніпуляції з рядками, валідація email тощо.

from enum import Enum
from pydantic import AnyUrl, BaseModel, EmailStr, Field


class MusicBand(str, Enum):
   AEROSMITH = "AEROSMITH"
   QUEEN = "QUEEN"
   ACDC = "AC/DC"


class UserBase(BaseModel):
    first_name: str = Field(min_length=1, max_length=128)
    username: str = Field(min_length=1, max_length=128, pattern="^[A-Za-z0-9-_]+$")
    email: EmailStr
    age: int = Field(ge=18, default=None)  # повинен бути більше або дорівнювати 18
    favorite_band: MusicBand | None = None  # дозволені лише значення "AEROSMITH", "QUEEN", "AC/DC"
    website: AnyUrl | None = None

Власна базова модель

Наявність контрольованої глобальної базової моделі дозволяє налаштовувати всі моделі в додатку. Наприклад, ми можемо встановити стандартний формат datetime або ввести спільний метод для всіх підкласів базової моделі.

from datetime import datetime
from zoneinfo import ZoneInfo

from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, ConfigDict


def datetime_to_gmt_str(dt: datetime) -> str:
    if not dt.tzinfo:
        dt = dt.replace(tzinfo=ZoneInfo("UTC"))

    return dt.strftime("%Y-%m-%dT%H:%M:%S%z")


class CustomModel(BaseModel):
    model_config = ConfigDict(
        json_encoders={datetime: datetime_to_gmt_str},
        populate_by_name=True,
    )

    def serializable_dict(self, **kwargs):
        """Повертає dict, який містить лише серіалізовані поля."""
        default_dict = self.model_dump()

        return jsonable_encoder(default_dict)

У наведеному прикладі ми вирішили створити глобальну базову модель, яка:

  • Серіалізує всі поля datetime у стандартний формат з явним часовим поясом
  • Надає метод для повернення dict лише з серіалізованими полями

Розділяйте Pydantic BaseSettings

BaseSettings був чудовим нововведенням для читання змінних середовища, але наявність одного BaseSettings для всього додатку може стати безладною з часом. Для покращення підтримуваності та організації ми розділили BaseSettings між різними модулями та доменами.

# src.auth.config
from datetime import timedelta

from pydantic_settings import BaseSettings


class AuthConfig(BaseSettings):
    JWT_ALG: str
    JWT_SECRET: str
    JWT_EXP: int = 5  # хвилини

    REFRESH_TOKEN_KEY: str
    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)

    SECURE_COOKIES: bool = True


auth_settings = AuthConfig()


# src.config
from pydantic import PostgresDsn, RedisDsn, model_validator
from pydantic_settings import BaseSettings

from src.constants import Environment


class Config(BaseSettings):
    DATABASE_URL: PostgresDsn
    REDIS_URL: RedisDsn

    SITE_DOMAIN: str = "myapp.com"

    ENVIRONMENT: Environment = Environment.PRODUCTION

    SENTRY_DSN: str | None = None

    CORS_ORIGINS: list[str]
    CORS_ORIGINS_REGEX: str | None = None
    CORS_HEADERS: list[str]

    APP_VERSION: str = "1.0"


settings = Config()

Залежності

Більше ніж Dependency Injection

Pydantic — чудовий валідатор схем, але для складних валідацій, які потребують виклику бази даних або зовнішніх сервісів, його недостатньо.

Документація FastAPI здебільшого представляє залежності як DI для ендпоінтів, але вони також чудово підходять для валідації запитів.

Залежності можуть використовуватися для валідації даних відповідно до обмежень бази даних (напр. перевірка існування email, перевірка наявності користувача тощо).

# dependencies.py
async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


# router.py
@router.get("/posts/{post_id}", response_model=PostResponse)
async def get_post_by_id(post: dict[str, Any] = Depends(valid_post_id)):
    return post


@router.put("/posts/{post_id}", response_model=PostResponse)
async def update_post(
    update_data: PostUpdate,
    post: dict[str, Any] = Depends(valid_post_id),
):
    updated_post = await service.update(id=post["id"], data=update_data)
    return updated_post


@router.get("/posts/{post_id}/reviews", response_model=list[ReviewsResponse])
async def get_post_reviews(post: dict[str, Any] = Depends(valid_post_id)):
    post_reviews = await reviews_service.get_by_post_id(post["id"])
    return post_reviews

Якби ми не помістили валідацію даних у залежність, нам довелося б валідувати існування post_id для кожного ендпоінту та писати однакові тести для кожного з них.

Ланцюжки залежностей

Залежності можуть використовувати інші залежності та уникати дублювання коду для подібної логіки.

# dependencies.py
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


async def parse_jwt_data(
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict[str, Any]:
    try:
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
    except JWTError:
        raise InvalidCredentials()

    return {"user_id": payload["id"]}


async def valid_owned_post(
    post: dict[str, Any] = Depends(valid_post_id),
    token_data: dict[str, Any] = Depends(parse_jwt_data),
) -> dict[str, Any]:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()

    return post

# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(post: dict[str, Any] = Depends(valid_owned_post)):
    return post

Розділяйте та повторно використовуйте залежності. Виклики залежностей кешуються

Залежності можуть повторно використовуватися кілька разів, і вони не будуть перераховуватися — FastAPI за замовчуванням кешує результат залежності в межах запиту, тобто якщо valid_post_id викликається кілька разів в одному маршруті, він буде викликаний лише один раз.

Знаючи це, ми можемо розділити залежності на кілька менших функцій, які працюють з меншою областю та легше повторно використовуються в інших маршрутах. Наприклад, у коді нижче ми використовуємо parse_jwt_data тричі:

  1. valid_owned_post
  2. valid_active_creator
  3. get_user_post,

але parse_jwt_data викликається лише один раз, при самому першому виклику.

# dependencies.py
from fastapi import BackgroundTasks
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

async def valid_post_id(post_id: UUID4) -> Mapping:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


async def parse_jwt_data(
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict:
    try:
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
    except JWTError:
        raise InvalidCredentials()

    return {"user_id": payload["id"]}


async def valid_owned_post(
    post: Mapping = Depends(valid_post_id),
    token_data: dict = Depends(parse_jwt_data),
) -> Mapping:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()

    return post


async def valid_active_creator(
    token_data: dict = Depends(parse_jwt_data),
):
    user = await users_service.get_by_id(token_data["user_id"])
    if not user["is_active"]:
        raise UserIsBanned()

    if not user["is_creator"]:
       raise UserNotCreator()

    return user


# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(
    worker: BackgroundTasks,
    post: Mapping = Depends(valid_owned_post),
    user: Mapping = Depends(valid_active_creator),
):
    """Отримати пост, що належить активному користувачу."""
    worker.add_task(notifications_service.send_email, user["id"])
    return post

Віддавайте перевагу async залежностям

FastAPI підтримує як sync, так і async залежності, і є спокуса використовувати sync залежності, коли не потрібно нічого await, але це може бути не найкращим вибором.

Так само, як і з маршрутами, sync залежності запускаються в пулі потоків. І потоки тут також мають свою ціну та обмеження, які є зайвими, якщо ви просто виконуєте невелику операцію без вводу/виводу.

Детальніше (зовнішнє посилання)

Різне

Дотримуйтесь REST

Розробка RESTful API спрощує повторне використання залежностей у маршрутах, таких як:

  1. GET /courses/:course_id
  2. GET /courses/:course_id/chapters/:chapter_id/lessons
  3. GET /chapters/:chapter_id

Єдине застереження — використовувати однакові назви змінних у шляху:

  • Якщо у вас є два ендпоінти GET /profiles/:profile_id та GET /creators/:creator_id, які обидва валідують існування profile_id, але GET /creators/:creator_id також перевіряє, чи профіль є creator, тоді краще перейменувати змінну шляху creator_id на profile_id та з'єднати ці дві залежності ланцюжком.
# src.profiles.dependencies
async def valid_profile_id(profile_id: UUID4) -> Mapping:
    profile = await service.get_by_id(profile_id)
    if not profile:
        raise ProfileNotFound()

    return profile

# src.creators.dependencies
async def valid_creator_id(profile: Mapping = Depends(valid_profile_id)) -> Mapping:
    if not profile["is_creator"]:
       raise ProfileNotCreator()

    return profile

# src.profiles.router.py
@router.get("/profiles/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(profile: Mapping = Depends(valid_profile_id)):
    """Отримати профіль за id."""
    return profile

# src.creators.router.py
@router.get("/creators/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(
     creator_profile: Mapping = Depends(valid_creator_id)
):
    """Отримати профіль creator за id."""
    return creator_profile

Серіалізація відповідей FastAPI

Якщо ви думаєте, що можете повернути Pydantic об'єкт, який відповідає response_model вашого маршруту, для оптимізації, то це неправильно.

FastAPI спочатку конвертує pydantic об'єкт у dict за допомогою jsonable_encoder, потім валідує дані з вашою response_model, і лише потім серіалізує об'єкт у JSON.

from fastapi import FastAPI
from pydantic import BaseModel, root_validator

app = FastAPI()


class ProfileResponse(BaseModel):
    @model_validator(mode="after")
    def debug_usage(self):
        print("created pydantic model")

        return self


@app.get("/", response_model=ProfileResponse)
async def root():
    return ProfileResponse()

Вивід логів:

[INFO] [2022-08-28 12:00:00.000000] created pydantic model
[INFO] [2022-08-28 12:00:00.000020] created pydantic model

Якщо потрібно використовувати синхронний SDK, запускайте його в пулі потоків.

Якщо вам необхідно використовувати бібліотеку для взаємодії з зовнішніми сервісами, і вона не async, виконуйте HTTP-виклики у зовнішньому робочому потоці.

Ми можемо використовувати відомий run_in_threadpool зі starlette.

from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from my_sync_library import SyncAPIClient

app = FastAPI()


@app.get("/")
async def call_my_sync_library():
    my_data = await service.get_my_data()

    client = SyncAPIClient()
    await run_in_threadpool(client.make_request, data=my_data)

ValueError може стати Pydantic ValidationError

Якщо ви викликаєте ValueError у схемі Pydantic, яка безпосередньо взаємодіє з клієнтом, він поверне гарну детальну відповідь користувачам.

# src.profiles.schemas
from pydantic import BaseModel, field_validator

class ProfileCreate(BaseModel):
    username: str

    @field_validator("password", mode="after")
    @classmethod
    def valid_password(cls, password: str) -> str:
        if not re.match(STRONG_PASSWORD_PATTERN, password):
            raise ValueError(
                "Password must contain at least "
                "one lower character, "
                "one upper character, "
                "digit or "
                "special symbol"
            )

        return password


# src.profiles.routes
from fastapi import APIRouter

router = APIRouter()


@router.post("/profiles")
async def get_creator_posts(profile_data: ProfileCreate):
   pass

Приклад відповіді:

Документація

  1. Якщо ваш API не публічний, приховуйте документацію за замовчуванням. Показуйте її явно лише на обраних середовищах.
from fastapi import FastAPI
from starlette.config import Config

config = Config(".env")  # парсинг .env файлу для змінних середовища

ENVIRONMENT = config("ENVIRONMENT")  # отримати назву поточного середовища
SHOW_DOCS_ENVIRONMENT = ("local", "staging")  # явний список дозволених середовищ

app_configs = {"title": "My Cool API"}
if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT:
   app_configs["openapi_url"] = None  # встановити url для документації як null

app = FastAPI(**app_configs)
  1. Допоможіть FastAPI генерувати зрозумілу документацію
  2. Встановлюйте response_model, status_code, description тощо.
  3. Якщо моделі та статуси відрізняються, використовуйте атрибут маршруту responses для додавання документації різних відповідей
from fastapi import APIRouter, status

router = APIRouter()

@router.post(
    "/endpoints",
    response_model=DefaultResponseModel,  # pydantic модель відповіді за замовчуванням
    status_code=status.HTTP_201_CREATED,  # код статусу за замовчуванням
    description="Description of the well documented endpoint",
    tags=["Endpoint Category"],
    summary="Summary of the Endpoint",
    responses={
        status.HTTP_200_OK: {
            "model": OkResponse, # власна pydantic модель для відповіді 200
            "description": "Ok Response",
        },
        status.HTTP_201_CREATED: {
            "model": CreatedResponse,  # власна pydantic модель для відповіді 201
            "description": "Creates something from user request ",
        },
        status.HTTP_202_ACCEPTED: {
            "model": AcceptedResponse,  # власна pydantic модель для відповіді 202
            "description": "Accepts request and handles it later",
        },
    },
)
async def documented_route():
    pass

Згенерує документацію ось так:

Встановіть конвенції іменування ключів БД

Явне встановлення іменування індексів відповідно до конвенцій вашої бази даних є кращим за автоматичне від sqlalchemy.

from sqlalchemy import MetaData

POSTGRES_INDEXES_NAMING_CONVENTION = {
    "ix": "%(column_0_label)s_idx",
    "uq": "%(table_name)s_%(column_0_name)s_key",
    "ck": "%(table_name)s_%(constraint_name)s_check",
    "fk": "%(table_name)s_%(column_0_name)s_fkey",
    "pk": "%(table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)

Міграції. Alembic

  1. Міграції повинні бути статичними та оборотними. Якщо ваші міграції залежать від динамічно згенерованих даних, переконайтеся, що єдине, що є динамічним — це самі дані, а не їхня структура.
  2. Генеруйте міграції з описовими назвами та slug. Slug обов'язковий і повинен пояснювати зміни.
  3. Встановіть зрозумілий шаблон файлу для нових міграцій. Ми використовуємо паттерн *date*_*slug*.py, напр. 2022-08-24_post_content_idx.py
# alembic.ini
file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s

Встановіть конвенції іменування БД

Послідовність у назвах важлива. Деякі правила, яких ми дотримувалися:

  1. lower_case_snake
  2. однина (напр. post, post_like, user_playlist)
  3. групуйте подібні таблиці з префіксом модуля, напр. payment_account, payment_bill, post, post_like
  4. будьте послідовними між таблицями, але конкретні назви допустимі, напр.
  5. використовуйте profile_id у всіх таблицях, але якщо деяким потрібні лише профілі, що є creators, використовуйте creator_id
  6. використовуйте post_id для всіх абстрактних таблиць як post_like, post_view, але конкретне іменування у відповідних модулях як course_id в chapters.course_id
  7. суфікс _at для datetime
  8. суфікс _date для date

Спочатку SQL. Потім Pydantic

  • Зазвичай база даних обробляє дані набагато швидше та чистіше, ніж CPython коли-небудь зможе.
  • Краще виконувати всі складні join та прості маніпуляції з даними за допомогою SQL.
  • Краще агрегувати JSON у БД для відповідей з вкладеними об'єктами.
# src.posts.service
from typing import Any

from pydantic import UUID4
from sqlalchemy import desc, func, select, text
from sqlalchemy.sql.functions import coalesce

from src.database import database, posts, profiles, post_review, products

async def get_posts(
    creator_id: UUID4, *, limit: int = 10, offset: int = 0
) -> list[dict[str, Any]]:
    select_query = (
        select(
            (
                posts.c.id,
                posts.c.slug,
                posts.c.title,
                func.json_build_object(
                   text("'id', profiles.id"),
                   text("'first_name', profiles.first_name"),
                   text("'last_name', profiles.last_name"),
                   text("'username', profiles.username"),
                ).label("creator"),
            )
        )
        .select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
        .where(posts.c.owner_id == creator_id)
        .limit(limit)
        .offset(offset)
        .group_by(
            posts.c.id,
            posts.c.type,
            posts.c.slug,
            posts.c.title,
            profiles.c.id,
            profiles.c.first_name,
            profiles.c.last_name,
            profiles.c.username,
            profiles.c.avatar,
        )
        .order_by(
            desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
        )
    )

    return await database.fetch_all(select_query)

# src.posts.schemas
from typing import Any

from pydantic import BaseModel, UUID4


class Creator(BaseModel):
    id: UUID4
    first_name: str
    last_name: str
    username: str


class Post(BaseModel):
    id: UUID4
    slug: str
    title: str
    creator: Creator


# src.posts.router
from fastapi import APIRouter, Depends

router = APIRouter()


@router.get("/creators/{creator_id}/posts", response_model=list[Post])
async def get_creator_posts(creator: dict[str, Any] = Depends(valid_creator_id)):
   posts = await service.get_posts(creator["id"])

   return posts

Встановіть async тестовий клієнт з першого дня

Написання інтеграційних тестів з БД, скоріш за все, призведе до заплутаних помилок циклу подій у майбутньому. Встановіть асинхронний тестовий клієнт одразу, напр. httpx

import pytest
from async_asgi_testclient import TestClient

from src.main import app  # ініціалізований FastAPI додаток


@pytest.fixture
async def client() -> AsyncGenerator[TestClient, None]:
    host, port = "127.0.0.1", "9000"

    async with AsyncClient(transport=ASGITransport(app=app, client=(host, port)), base_url="http://test") as client:
        yield client


@pytest.mark.asyncio
async def test_create_post(client: TestClient):
    resp = await client.post("/posts")

    assert resp.status_code == 201

Якщо тільки у вас немає синхронних з'єднань з БД (серйозно?) або ви не плануєте писати інтеграційні тести.

Використовуйте ruff

З лінтерами ви можете забути про форматування коду і зосередитися на написанні бізнес-логіки.

Ruff — це "блискавично швидкий" новий лінтер, який замінює black, autoflake, isort та підтримує понад 600 правил.

Використання pre-commit hooks — поширена гарна практика, але для нас достатньо було просто використовувати скрипт.

#!/bin/sh -e
set -x

ruff check --fix src
ruff format src

Бонусний розділ

Деякі дуже добрі люди поділилися власним досвідом та найкращими практиками, які безумовно варто прочитати. Переглядайте їх у розділі issues проекту.

Наприклад, lowercase00 детально описав свої найкращі практики роботи з permissions та auth, сервісами та views на основі класів, чергами задач, власними серіалізаторами відповідей, конфігурацією з dynaconf тощо.

Якщо у вас є чим поділитися про ваш досвід роботи з FastAPI, чи то позитивний чи негативний, ви можете створити новий issue. Нам буде приємно прочитати.