Конфигурация, переменные окружения и архитектура приложения

⏱ 45 мин · 🧩 заданий: 4

Конфигурация через окружение

Один и тот же код запускается на ноутбуке, в тестах, на staging и в продакшене. Отличаются только настройки: адрес базы, ключи, режим отладки. Принцип 12-factor: конфигурация живёт в переменных окружения, а не в коде.

import os

os.environ["DATABASE_URL"] = "postgresql://shop:secret@db:5432/shop"   # обычно задаёт платформа, а не код
os.environ["DEBUG"] = "1"

database_url = os.environ["DATABASE_URL"]            # обязательная: KeyError, если нет — упадём сразу при старте
debug = os.getenv("DEBUG", "0") == "1"               # необязательная с умолчанием
workers = int(os.getenv("WORKERS", "4"))             # всё в окружении — строки, приводим сами
print(database_url, debug, workers)
print(bool("0"), bool("false"))                      # ловушка: непустая строка — всегда True

Для разработки переменные кладут в файл .env (он в .gitignore!), а в репозиторий — .env.example с перечнем переменных без настоящих значений:

# .env.example
DATABASE_URL=postgresql://shop:shop@localhost:5432/shop
REDIS_URL=redis://localhost:6379/0
SECRET_KEY=change-me
DEBUG=1
ALLOWED_HOSTS=localhost,127.0.0.1

Загружают .env библиотекой python-dotenv, docker compose (env_file:) или самим pydantic-settings.

pydantic-settings

Типизированные настройки с валидацией при старте: если переменной нет или значение неверное, приложение не запустится с понятной ошибкой, вместо того чтобы упасть через час на первом запросе.

# app/config.py
from functools import lru_cache
from pydantic import Field, PostgresDsn, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="SHOP_", extra="ignore")

    database_url: PostgresDsn
    redis_url: str = "redis://localhost:6379/0"
    secret_key: SecretStr                     # не печатается в логах: SecretStr('**********')
    debug: bool = False                       # "1", "true", "yes", "on" → True
    allowed_hosts: list[str] = ["localhost"]  # из JSON: SHOP_ALLOWED_HOSTS='["a.ru","b.ru"]'
    workers: int = Field(4, ge=1, le=64)

@lru_cache                                    # один экземпляр на процесс
def get_settings() -> Settings:
    return Settings()

В FastAPI настройки получают зависимостью Depends(get_settings), а в тестах подменяют. Похожий механизм на чистом Pydantic можно потрогать прямо здесь:

from pydantic import BaseModel, Field, SecretStr, ValidationError, field_validator

class Settings(BaseModel):
    database_url: str
    secret_key: SecretStr
    debug: bool = False
    allowed_hosts: list[str] = []
    workers: int = Field(4, ge=1, le=64)

    @field_validator("allowed_hosts", mode="before")
    @classmethod
    def split_csv(cls, v):
        return [h.strip() for h in v.split(",") if h.strip()] if isinstance(v, str) else v

env = {"DATABASE_URL": "postgresql://db/shop", "SECRET_KEY": "s3cr3t", "DEBUG": "yes", "ALLOWED_HOSTS": "a.ru, b.ru"}
settings = Settings(**{k.lower(): v for k, v in env.items()})
print(settings)
print(settings.secret_key.get_secret_value())

try:
    Settings(database_url="x", secret_key="k", debug="может быть", workers="100")
except ValidationError as e:
    print([f"{err['loc'][0]}: {err['msg']}" for err in e.errors()])

Окружения

  • local — DEBUG=1, локальная база в docker compose, подробные логи.
  • test — отдельная база (или SQLite в памяти), фиксированные ключи, внешние сервисы подменены.
  • staging — копия продакшена с тестовыми данными: проверка перед выкаткой.
  • production — DEBUG=0, секреты из менеджера секретов, логи в JSON, мониторинг.

Не плодите if ENV == "production" по всему коду: различия — только в значениях настроек.

pyproject.toml и раскладка проекта

shop/
├── pyproject.toml          # метаданные, зависимости, настройки инструментов
├── uv.lock / poetry.lock   # точные версии всех зависимостей (коммитится!)
├── .env.example
├── .gitignore
├── .pre-commit-config.yaml
├── Dockerfile
├── compose.yaml
├── README.md
├── src/
│   └── shop/
│       ├── __init__.py
│       ├── main.py
│       ├── config.py
│       ├── api/
│       ├── services/
│       ├── repositories/
│       ├── models/
│       └── schemas/
├── migrations/
└── tests/
    ├── conftest.py
    ├── unit/
    └── integration/
[project]
name = "shop"
version = "1.4.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115",
    "sqlalchemy>=2.0",
    "pydantic-settings>=2.4",
    "psycopg[binary]>=3.2",
]

[project.optional-dependencies]
dev = ["pytest>=8", "pytest-cov", "mypy", "ruff", "pre-commit"]

[tool.ruff]
line-length = 100
[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "SIM"]   # pycodestyle, pyflakes, isort, bugbear, pyupgrade…

[tool.mypy]
strict = true

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-q --cov=shop"

src-раскладка (src/shop/) гарантирует, что тесты импортируют установленный пакет, а не случайно файлы из текущей папки.

Зависимости и lock-файлы

В pyproject.toml пишут диапазоны (fastapi>=0.115) — что нужно приложению. В lock-файле — точные версии всех пакетов, включая транзитивные, с хешами: сборка воспроизводима, и завтра CI поставит ровно то же самое.

Инструмент Как работает
pip + requirements.txt вручную; pip freeze > requirements.txt фиксирует всё подряд
pip-tools requirements.in → pip-compile → requirements.txt с точными версиями
Poetry pyproject.toml + poetry.lock, своё виртуальное окружение, публикация пакетов
uv очень быстрая замена pip/pip-tools/venv/pyenv: uv add fastapi, uv sync, uv run pytest, uv.lock

Обновляйте зависимости регулярно (Dependabot/Renovate) и проверяйте уязвимости (pip-audit).

Архитектура: слои

HTTP-запрос │ ▼ API (роутеры FastAPI / views Django) разбор запроса, коды ответа, сериализация │ вызывает ▼ Сервисы (бизнес-логика) правила предметной области, транзакции │ использует ▼ Репозитории / ORM запросы к базе │ ▼ PostgreSQL, Redis, внешние API

Зависимости направлены внутрь: сервис не знает про HTTP, поэтому его можно вызвать из API, из CLI-команды, из фоновой задачи и легко протестировать.

from dataclasses import dataclass
from typing import Protocol

# --- домен ---
class DomainError(Exception):
    code = "domain_error"

class NotEnoughStock(DomainError):
    code = "not_enough_stock"

@dataclass
class Product:
    id: int
    title: str
    price: int
    stock: int

# --- порт: что сервису нужно от хранилища ---
class ProductRepository(Protocol):
    def get(self, product_id: int) -> Product | None: ...
    def save(self, product: Product) -> None: ...

# --- сервис: бизнес-логика, не знает ни про HTTP, ни про SQL ---
class OrderService:
    def __init__(self, products: ProductRepository, discount_percent: int = 0):
        self.products = products
        self.discount_percent = discount_percent

    def place_order(self, product_id: int, quantity: int) -> int:
        product = self.products.get(product_id)
        if product is None:
            raise LookupError(f"товар {product_id} не найден")
        if product.stock < quantity:
            raise NotEnoughStock(f"осталось {product.stock} шт.")
        product.stock -= quantity
        self.products.save(product)
        return product.price * quantity * (100 - self.discount_percent) // 100

# --- адаптер: реализация для тестов (в проде — на SQLAlchemy) ---
class InMemoryProducts:
    def __init__(self, *items: Product):
        self.items = {p.id: p for p in items}
    def get(self, product_id):
        return self.items.get(product_id)
    def save(self, product):
        self.items[product.id] = product

# --- «API»: переводит исключения домена в HTTP-коды ---
def handle_post_order(service: OrderService, payload: dict) -> tuple[int, dict]:
    try:
        total = service.place_order(payload["product_id"], payload["quantity"])
    except NotEnoughStock as e:
        return 409, {"error": e.code, "message": str(e)}
    except LookupError as e:
        return 404, {"error": "not_found", "message": str(e)}
    return 201, {"total": total}

repo = InMemoryProducts(Product(1, "Кружка", 500, stock=3))
service = OrderService(repo, discount_percent=10)          # внедрение зависимостей через конструктор
print(handle_post_order(service, {"product_id": 1, "quantity": 2}))
print(handle_post_order(service, {"product_id": 1, "quantity": 2}))
print(handle_post_order(service, {"product_id": 7, "quantity": 1}))

Не превращайте это в догму: в маленьком CRUD-сервисе отдельные репозитории и интерфейсы — лишний код. Слои окупаются, когда растёт бизнес-логика и команда.

Кеширование и Redis

Redis — хранилище «ключ — значение» в памяти. Применения: кеш тяжёлых запросов, сессии, счётчики и rate limiting, блокировки, брокер очередей.

import json
import redis

r = redis.Redis.from_url(settings.redis_url, decode_responses=True)

def get_product(product_id: int) -> dict:
    key = f"product:{product_id}"
    if (cached := r.get(key)) is not None:
        return json.loads(cached)                    # попадание в кеш
    product = repo.load(product_id)                  # промах — идём в базу
    r.set(key, json.dumps(product), ex=300)          # TTL 5 минут
    return product

def update_product(product_id: int, data: dict):
    repo.update(product_id, data)
    r.delete(f"product:{product_id}")                # инвалидация кеша

«В информатике две сложные вещи: инвалидация кеша и придумывание имён». Всегда задавайте TTL и продумайте, когда кеш сбрасывать. Внутри одного процесса хватает functools.lru_cache / cache.

Фоновые задачи и очереди

Долгую работу (письма, отчёты, обработка файлов, вызовы медленных API) не делают внутри HTTP-запроса. Её кладут в очередь, а выполняют отдельные процессы-воркеры: Celery (с Redis или RabbitMQ), RQ, arq, Dramatiq. Задачи должны быть идемпотентными: при сбое очередь может выполнить задачу повторно.

# tasks.py (Celery)
@app.task(bind=True, max_retries=5, autoretry_for=(ConnectionError,), retry_backoff=True)
def send_receipt(self, order_id: int):
    order = Order.get(order_id)
    if order.receipt_sent:          # идемпотентность: повтор ничего не сломает
        return
    mailer.send(order.email, render_receipt(order))
    order.mark_receipt_sent()

# в обработчике запроса
send_receipt.delay(order.id)        # передаём id, а не объект

Периодические задачи — Celery beat, cron, планировщик платформы.

Наблюдаемость

  • Логи — структурированные (JSON) с request_id, уровнем и контекстом; в stdout. Не логируйте пароли, токены и персональные данные.
  • Метрики — числа во времени: RPS, время ответа (p50/p95/p99), доля ошибок, длина очередей. Prometheus + Grafana.
  • Трассировка — путь одного запроса через сервисы (OpenTelemetry, Jaeger).
  • Ошибки — Sentry: стектрейс, контекст, группировка, уведомления.
  • Алерты — на симптомы, которые видит пользователь (ошибки, задержки), а не на каждый скачок CPU.

Итог

  • Конфигурация — в переменных окружения; .env для разработки, .env.example в репозитории, секреты — никогда в коде и образах.
  • Типизированные настройки валидируются при старте (pydantic-settings).
  • pyproject.toml + lock-файл; ruff, mypy, pytest, pre-commit в CI.
  • Слои: API → сервисы → репозитории; зависимости внедряются, домен не знает про HTTP.
  • Кеш с TTL и инвалидацией, долгие задачи — в очередь с идемпотентными воркерами, логи/метрики/трассировка/Sentry.

Практика решено 0 из 4

Вы не вошли — ответы проверяются, но прогресс не сохранится. Войти или зарегистрироваться.
Задание 1 Программа на Python

Парсер .env. Напишите parse_env(text), возвращающую словарь переменных из содержимого .env-файла:

  • строки КЛЮЧ=значение; пробелы вокруг ключа и значения обрезаются; допускается префикс export;
  • пустые строки и строки, начинающиеся с #, пропускаются; комментарий # после значения без кавычек отбрасывается;
  • значение в двойных или одинарных кавычках берётся как есть (без кавычек, # внутри — часть значения); в двойных кавычках \n превращается в перевод строки;
  • строка без = — ошибка: ValueError с номером строки (считая с 1) в сообщении.
Какие тесты проверяют код (3)
text = """
# настройки разработки
DEBUG=1
DATABASE_URL = postgresql://shop:shop@localhost:5432/shop
export SECRET_KEY=abc=def==   # base64 с «=» в конце
EMPTY=
"""
assert parse_env(text) == {"DEBUG": "1", "DATABASE_URL": "postgresql://shop:shop@localhost:5432/shop",
                           "SECRET_KEY": "abc=def==", "EMPTY": ""}

text = 'GREETING="Привет, # мир"\nSINGLE=\'a\\nb\'\nMULTI="строка1\\nстрока2"'
env = parse_env(text)
assert env["GREETING"] == "Привет, # мир" and env["SINGLE"] == "a\\nb" and env["MULTI"] == "строка1\nстрока2"

try:
    parse_env("A=1\n\nоops")
except ValueError as e:
    assert "3" in str(e), "В сообщении нужен номер строки"
else:
    raise AssertionError("Строка без = должна вызывать ValueError")

Задание 2 Программа на Python

Типизированные настройки. Напишите load_settings(env), которая собирает из словаря переменных окружения env объект Settings (dataclass уже описан в заготовке) и проверяет значения:

  • DATABASE_URL — обязательна; если нет — ValueError, в сообщении имя переменной;
  • DEBUG — булево: 1/true/yes/on → True, 0/false/no/off или отсутствует → False (регистр неважен), другое значение — ValueError;
  • PORT — целое от 1 до 65535, по умолчанию 8000;
  • ALLOWED_HOSTS — список через запятую, пробелы обрезаются, пустые элементы выбрасываются; по умолчанию [].

Так работает pydantic-settings, а вы напишете его упрощённую версию.

Какие тесты проверяют код (2)
s = load_settings({"DATABASE_URL": "sqlite:///db.sqlite3"})
assert s == Settings("sqlite:///db.sqlite3", False, 8000, [])
s = load_settings({"DATABASE_URL": "pg://x", "DEBUG": "Yes", "PORT": "9000", "ALLOWED_HOSTS": " a.ru, b.ru ,,"})
assert s == Settings("pg://x", True, 9000, ["a.ru", "b.ru"])
assert load_settings({"DATABASE_URL": "pg://x", "DEBUG": "0"}).debug is False
assert load_settings({"DATABASE_URL": "pg://x", "DEBUG": "false"}).debug is False

for env, name in (({}, "DATABASE_URL"), ({"DATABASE_URL": "x", "DEBUG": "maybe"}, "DEBUG"),
                  ({"DATABASE_URL": "x", "PORT": "abc"}, "PORT"), ({"DATABASE_URL": "x", "PORT": "70000"}, "PORT")):
    try:
        load_settings(env)
    except ValueError as e:
        assert name in str(e), f"В сообщении об ошибке нужно имя переменной {name}"
    else:
        raise AssertionError(f"{env} должен вызывать ValueError")

Задание 3 Один вариант ответа

Где хранить пароль к продакшен-базе?

Задание 4 Заполнить таблицу

К какому слою относится код?

API / обработчикисервисы (бизнес-логика)репозитории (доступ к данным)
Разобрать JSON запроса и вернуть 201 с телом ответа
Проверить, что на складе хватает товара, рассчитать скидку и создать заказ
SELECT заказов пользователя за период
Решить, можно ли отменить заказ в статусе «отправлен»
Отметьте ячейку — «+», пустая — «−».

Войдите, чтобы отмечать пройденные уроки.