Конфигурация, переменные окружения и архитектура приложения
Конфигурация через окружение
Один и тот же код запускается на ноутбуке, в тестах, на 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, из 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
Парсер .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")
def parse_env(text):
result = {}
for number, raw in enumerate(text.splitlines(), start=1):
line = raw.strip()
if not line or line.startswith("#"):
continue
if line.startswith("export "):
line = line[len("export "):]
key, sep, value = line.partition("=")
if not sep:
raise ValueError(f"строка {number}: ожидается КЛЮЧ=значение")
key, value = key.strip(), value.strip()
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
quote, value = value[0], value[1:-1]
if quote == '"':
value = value.replace("\\n", "\n")
else:
value = value.split(" #", 1)[0].strip()
result[key] = value
return result
Типизированные настройки. Напишите 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")
from dataclasses import dataclass, field
@dataclass(frozen=True)
class Settings:
database_url: str
debug: bool = False
port: int = 8000
allowed_hosts: list[str] = field(default_factory=list)
TRUE, FALSE = {"1", "true", "yes", "on"}, {"0", "false", "no", "off"}
def _bool(name, value):
v = value.strip().lower()
if v in TRUE:
return True
if v in FALSE:
return False
raise ValueError(f"{name}: ожидается булево значение, получено {value!r}")
def load_settings(env):
if not env.get("DATABASE_URL"):
raise ValueError("DATABASE_URL: обязательная переменная")
try:
port = int(env.get("PORT", "8000"))
except ValueError:
raise ValueError("PORT: ожидается целое число") from None
if not 1 <= port <= 65535:
raise ValueError("PORT: от 1 до 65535")
hosts = [h.strip() for h in env.get("ALLOWED_HOSTS", "").split(",") if h.strip()]
return Settings(
database_url=env["DATABASE_URL"],
debug=_bool("DEBUG", env.get("DEBUG", "false")),
port=port,
allowed_hosts=hosts,
)
Где хранить пароль к продакшен-базе?
К какому слою относится код?
| API / обработчики | сервисы (бизнес-логика) | репозитории (доступ к данным) | |
|---|---|---|---|
| Разобрать JSON запроса и вернуть 201 с телом ответа | |||
| Проверить, что на складе хватает товара, рассчитать скидку и создать заказ | |||
| SELECT заказов пользователя за период | |||
| Решить, можно ли отменить заказ в статусе «отправлен» |
Войдите, чтобы отмечать пройденные уроки.