Конфігурування FastAPI за допомогою ConfZ

Confz — це нова бібліотека керування конфігурацією для Python. Вона базується на pydantic, що робить її ідеальною для використання разом з FastAPI. У поєднанні з новою SQLModel ORM, що також використовує pydantic під капотом, ви отримуєте ідеальне тріо для вашого API.

Спочатку розглянемо приклад проєкту, що складається з FastAPI та SQLModel. Потім ми додамо ConfZ і покажемо, як легко ми можемо:

• Сконфігурувати API без хардкодингу;
• Керувати базами даних для всіх середовищ (dev, test, prod, …);
• Писати тести без необхідності вкраплювати будь-який код.

Весь код доступний на GitHub.

Приклад проєкту

Почнімо з прикладу проєкту на основі FastAPI і SQLModel, натхненного документацією SQLModel. Цей API дозволяє керувати користувачами. Наразі користувач складається лише з id та name. Як рекомендовано в документації, ми створюємо клас UserBase з усіма загальними полями, а потім додаємо специфікації для таблиці, сценарії читання та створення (на цю мить оновлення даних не враховуються).

from typing import Optional
from sqlmodel import Field, SQLModel

class UserBase(SQLModel):
    name: str

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class UserCreate(UserBase):
    pass

class UserRead(UserBase):
    id: int

Нам також потрібна певна логіка, щоб створити механізм і керувати сеансами:

from sqlmodel import SQLModel, Session, create_engine

engine = create_engine("sqlite:///dev_db.db", echo=True, connect_args={"check_same_thread": False})

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

І звісно ми потребуємо свій API:

from confz import validate_all_configs
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi_confz_demo.db import create_db_and_tables

app = FastAPI(title="My API", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost",
        "https://my-domain.com"
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

Тепер ми маємо всі складники для створення наших маршрутів:

from typing import List
from fastapi import Depends
from sqlmodel import Session, select
from fastapi_confz_demo.db import get_session
from fastapi_confz_demo.models import User, UserRead, UserCreate

@app.post("/user/", response_model=UserRead)
def create_user(*, session: Session = Depends(get_session), user: UserCreate):
    db_user = User.from_orm(user)
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

@app.get("/user/", response_model=List[UserRead])
def read_users(*, session: Session = Depends(get_session)):
    users = session.exec(select(User)).all()
    return users

Готово. Завдяки акуратній інтеграції FastAPI та SQLModel нам потрібні лише ці кілька рядків коду для (базового) API керування користувачами. Для простоти ми ігноруємо оновлення та видалення в цьому прикладі (але вони також будуть дуже простими).

Конфігурація нашого API

У наш файлі API ми захардкодили певну конфігурацію, як-от назва API або джерела CORS. У реальному проєкті так робити не варто. Краще помістити їх, наприклад, у файл конфігурації yaml:

title: "My API"
version: "1.0.0"
cors_origins:
  - "http://localhost"
  - "https://my-domain.com"

Завдяки ConfZ можна завиграшки завантажувати цей файл, перевіряти його вміст і потім використовувати його в нашому застосунку. Нам просто потрібно визначити схему конфігурації (так само як ми визначаємо схему API або схему БД) і вказати розташування файлу конфігурації:

from pathlib import Path
from typing import List
from confz import ConfZ, ConfZFileSource
from pydantic import AnyUrl

CONFIG_DIR = Path(__file__).parent.parent.resolve() / "config"

class AppConfig(ConfZ):
    title: str
    version: str
    cors_origins: List[AnyUrl]
    CONFIG_SOURCES = ConfZFileSource(file=CONFIG_DIR / "api.yml")

Тепер ми можемо використовувати цю конфігурацію безпосередньо без будь-яких додаткових дій:

from fastapi_confz_demo.config import AppConfig

app = FastAPI(title=AppConfig().title, version=AppConfig().version)

app.add_middleware(
    CORSMiddleware,
    allow_origins=AppConfig().cors_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

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

Конфігурація нашої бази даних

Хоча наша конфігурація API відносно статична, конфігурація БД залежить від середовища:

• Під час розробки ми використовуємо локальну базу даних SQLite для зручності;
• У робочому середовищі ми використовуємо зовнішню базу даних PostgreSQL;
• Для тестування ми використовуємо кешовану базу даних SQLite.

Для максимальної гнучкості добре б мати файл конфігурації для кожного середовища (крім тестування, яке ми розглянемо пізніше). Щоб розрізняти різні середовища нам потрібно прочитати відповідну змінну середовища. Будь-які облікові дані бази даних також слід читати з середовища, щоб не розміщувати їх безпосередньо у файлах конфігурації

Хоча на перший погляд ці вимоги можуть здатися складними для реалізації, насправді це дуже просто за допомогою ConfZ. Спочатку ми визначаємо наші класи конфігурації та джерела:

from pathlib import Path
from typing import Union, Optional, Literal
from confz import ConfZ, ConfZFileSource, ConfZEnvSource
from pydantic import SecretStr

CONFIG_DIR = Path(__file__).parent.parent.resolve() / "config"

class SQLiteDB(ConfZ):
    type: Literal["sqlite"]
    path: Optional[Path]  # None if in-memory

class PostgreSQL(ConfZ):
    type: Literal["postgresql"]
    user: str
    password: SecretStr
    host: str
    database: str

DBTypes = Union[SQLiteDB, PostgreSQL]

class DBConfig(ConfZ):
    echo: bool
    db: DBTypes
    CONFIG_SOURCES = [
        ConfZFileSource(
            folder=CONFIG_DIR,
            file_from_env="DB_ENV"
        ),
        ConfZEnvSource(allow=[
            "db.user",
            "db.password"
        ])
    ]

Конфігурація нашої бази даних DBConfig містить echo-налаштування та два типи баз даних. Залежно від поля type, це або база даних SQLite (з кешу або у вказаного місця) або база даних PostgreSQL. CONFIG_SOURCES визначає, що ConfZ має перевірити теку конфігурації та використовувати файл, який вказаний у змінній середовищі DB_ENV. Ми можемо встановити цю змінну до dbdev.yml з файлом:

echo: True
db:
  type: "sqlite"
  path: "dev_db.db"

або з файлом dbprod.yml:

В останньому випадку нам також доведеться встановити змінні середовища DB.USER і DB.PASSWORD.

Тепер ми можемо використовувати цю конфігурацію, щоб створити нашу базу даних:

from sqlalchemy.pool import StaticPool
from sqlmodel import create_engine
from fastapi_confz_demo.config import DBTypes, SQLiteDB, PostgreSQL, DBConfig

def get_db_args(db: DBTypes):
    if isinstance(db, SQLiteDB):
        connect_args = {"check_same_thread": False}
        if db.path is None:
            url = "sqlite://"
            args = {"connect_args": connect_args, "poolclass": StaticPool}
        else:
            url = f"sqlite:///{db.path}"
            args = {"connect_args": connect_args}

    elif isinstance(db, PostgreSQL):
        url = f"postgresql://{db.user}:{db.password.get_secret_value()}@{db.host}/{db.database}"
        args = {}
    else:
        raise ValueError(f"Invalid DB type '{type(db)}'.")
				
    return url, args

_url, _args = get_db_args(DBConfig().db)
engine = create_engine(_url, echo=DBConfig().echo, **_args)

Ось і все, ми охопили всі вимоги за допомогою цих кількох рядків коду.

Тестування

FastAPI має розвинену систему додавання залежностей для полегшення тестування. Однак у цьому навчальному прикладі нам це не потрібно, тому що наш клас бази даних вже підтримує базу даних в пам'яті, нам просто потрібно адаптувати конфігурацію та встановити db.path на None. Кожен клас конфігурації в ConfZ підтримує це з коробки за допомогою спеціального контекстного менеджера:

import pytest
from confz import ConfZDataSource
from fastapi_confz_demo.app import on_startup
from fastapi_confz_demo.config import DBConfig

@pytest.fixture(name="db", autouse=True)
def db_fixture():
    new_sources = ConfZDataSource(data={
        "echo": True,
        "db": {
            "type": "sqlite",
            "path": None
        }
    })

    with DBConfig.change_config_sources(new_sources):
        on_startup()
        yield

Те, що ми визначаємо тут, — це закріплена база даних, яка має autouse=True, тож ми ніколи випадково не доступимося до розробницької (або, що гірше, до робочої) бази даних. Ми вказуємо конфігурацію для тестування безпосередньо в коді та перезаписуємо наявне джерело конфігурації на нове.

Тут дечого бракує: ми визначили engine на рівні модуля, тому зміна конфігурації не допоможе нам безпосередньо, оскільки engine уже створено. Щоб запобігти цьому, ми замінюємо безпосереднє створення функцією геттер:

from confz import depends_on

from fastapi_confz_demo.config import DBConfig

@depends_on(DBConfig)
def get_engine():
    url, args = get_db_args(DBConfig().db)
    engine = create_engine(url, echo=DBConfig().echo, **args)
    return engine

Декоратор depends_on від ConfZ гарантує, що ми повертаємо кешовану копію після першого виклику. Він також гарантує, що коли ми змінюємо DBConfig за допомогою нашого контекстного менеджера, ми повторно створюємо engine.

Тепер ми можемо замінити всі доступи до engine на get_engine(), і ConfZ знатиме, що всі маршрути FastAPI використовують новий engine.

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

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session
from fastapi_confz_demo.app import app, User, on_startup
from fastapi_confz_demo.db import get_engine

@pytest.fixture(name="session")
def session_fixture():
    with Session(get_engine()) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture():
    return TestClient(app)

def test_create_user(session: Session, client: TestClient):
    response = client.post("/user/", json={"name": "my-name"})
    data = response.json()
    assert response.status_code == 200
    assert data["name"] == "my-name"
    assert data["id"] is not None

    user = session.get(User, data["id"])
    assert user is not None
    assert user.name == "my-name"

def test_create_user_incomplete(client: TestClient):
    response = client.post("/user/", json={"not-needed": "empty"})
    assert response.status_code == 422

def test_create_user_invalid(client: TestClient):
    response = client.post("/user/", json={"name": {"something": "useless"}})
    assert response.status_code == 422

def test_read_users(session: Session, client: TestClient):
    user_1 = User(name="user1")
    user_2 = User(name="user2")
    session.add(user_1)
    session.add(user_2)
    session.commit()

    response = client.get("/user/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == user_1.name
    assert data[0]["id"] == user_1.id
    assert data[1]["name"] == user_2.name
    assert data[1]["id"] == user_2.id

Рання перевірка

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

Це відбувається у той час, коли наша конфігурація ще не вказана коректно (наприклад, змінна DB_ENV не призначена) і ConfZ поверне помилку. Це може бути досить запізніло, і було б правильно перевірити все на початку запуску. На щастя, ConfZ також має рішення для цього. Нам просто потрібно налаштувати наш метод on_startup:

from confz import validate_all_configs

@app.on_event("startup")
def on_startup():
    validate_all_configs(include_listeners=True)
    create_db_and_tables()

validate_all_configs пройде через усі класи конфігурації, завантажуючи їхні джерела та перевіряючи вміст. Таким чином, помилка в конфігурації бази даних вже буде виявлена на той час. Прапор include_listeners впевнюється, що get_engine також уже виконується, тому конфігурація має не лише правильну схему, але також уможливлює під'єднання до нашої бази даних.

Висновок

Як ми побачили, ConfZ дуже просто інтегрується з FastAPI та SQLModel. Разом вони утворюють тріо, яке можна використовувати для легкого створення добре конфігурованих і потужних API.

ConfZ доступний на GitHub. Тож якщо хочете, можете спробувати погратися з ним.