Несколько месяцев я делал pet-проект, который зашёл дальше, чем планировалось: файловый ASGI-фреймворк на Python под названием EndoCore. Сегодня вышла версия 1.0.0, и это хороший повод рассказать, что это, зачем, и что было больнее всего сделать правильно.

Сразу ссылки, чтобы не листать до конца:

• PyPI: https://pypi.org/project/endocore/

• GitHub: https://github.com/Drakulyonok/endocore

• Документация (EN/RU): https://endocore.readthedocs.io

• pip install endocore

Любой растущий API-проект на декораторах рано или поздно расходится сам с собой: таблица роутов говорит одно, хендлеры — другое, а вопрос “к какой версии относится этот эндпоинт” превращается в археологию. С ростом становится только хуже — роуты живут в голове у того, кто их писал, разбросаны по файлам, которые импортируют друг друга в произвольном порядке.

Я захотел эту дрейфующую сущность просто убрать. Не “уменьшить риск”, а сделать так, чтобы дрейфовать было физически нечему.

Api/v1/User/[id]/Get.py    ->  GET  /v1/user/42        (id="42")
Api/v1/User/Role/Post.py   ->  POST /v1/user/role
Api/v2/User/[id]/Get.py    ->  GET  /v2/user/42        (v1 продолжает работать, нетронут)

Кладёшь файл в нужную папку — эндпоинт существует: заведён, версионирован, показывается в endo routes и /docs, без единой строчки регистрации. Удаляешь файл — эндпоинт исчезает. Никакого отдельного роутера, который может разойтись с тем, что реально делает код, потому что отдельного роутера просто нет — дерево читается напрямую.

# Api/v1/User/Role/Post.py  ->  POST /v1/user/role
from endocore import Request, Response

async def handler(request: Request) -> Response:
    data = await request.json()
    return Response.json({"created": data["name"]}, status=201)

Это уже полноценный рабочий эндпоинт. Не app = FastAPI(), не @app.post(...), никакого импорта, который надо было бы куда-то подключить. Путь и имя файла — это весь контракт.

Версионирование в этой модели становится тривиальным: v2 — это shutil.copytree с фильтром. v1 не шарит состояние роутера с v2 и не может быть задет его изменением. Никаких if version == 2 в хендлерах, никакого версионирования, которое работает только если все помнят конвенцию.

Один pip install, один процесс, ничего собирать руками:

• ORM — SQLite и PostgreSQL, синхронный и асинхронный API, пул соединений, миграции с откатом. С 1.0 можно опционально включить нативный async на Postgres (async_native=True) — это не threadpool-обёртка над синхронным движком, а честный AsyncConnection из psycopg3, строго по желанию, поведение существующего деплоя не меняется при апгрейде.

• Безопасность — только параметризованный SQL, идентификаторы валидируются и квотятся, пароли — через scrypt, подписанные сессии, CSRF, rate limiting.

• Реалтайм — файловые WebSocket’ы (Socket.py) + pub/sub комнаты, которые можно разнести по воркерам через Redis fan-out.

• DI — Depends(...) в духе FastAPI, вложенный, кэшируется на запрос.

• Тестирование — TestClient, добавленный специально для 1.0: внутрипроцессный ASGI-клиент без сетевого сокета и без лишней зависимости, драйвит и HTTP, и WebSocket-сессии.

• Наблюдаемость — структурированное логирование с маскировкой секретов, Prometheus-метрики, OpenTelemetry-трейсинг, /openapi.json + Swagger UI.

• Интеграции — Redis, Celery, SMTP — через extensions.py.

Обязательная зависимость всего одна — uvicorn. Резолвер, загрузчик, Request/Response, цепочка middleware, ORM и CLI — всё на стандартной библиотеке.

from endocore.orm import Model, fields, configure, create_all, Q, F

class User(Model):
    name   = fields.CharField(max_length=100)
    age    = fields.IntegerField(default=0)
    active = fields.BooleanField(default=True)

configure(backend="sqlite", database="app.db")   # или backend="postgres", pool_size=10, ...
create_all(User)

User.objects.create(name="Ada", age=36)
User.objects.filter(age__gte=18).order_by("-age")             # ленивый QuerySet
User.objects.filter(Q(age__lt=18) | Q(name__icontains="a"))   # Q-объекты
User.objects.filter(age__gte=18).update(active=True)          # bulk update
User.objects.filter(pk=1).update(age=F("age") + 1)            # атомарный F()-expression

# неблокирующий вызов для ASGI-хендлеров:
user = await User.objects.aget(pk=1)

Каждое значение биндится через драйвер (никогда не форматируется строкой в SQL), каждый идентификатор валидируется и квотится, в SQL превращается только фиксированный whitelist лукапов, LIMIT/OFFSET принудительно приводятся к int. Это не опциональный слой — это единственный способ, которым ORM вообще умеет строить запрос.

В какой-то момент я перестал добавлять фичи и целый релиз (0.9.0b1) потратил на то, чтобы целенаправленно ломать собственный фреймворк — не читать код в поисках подозрительных мест, а воспроизводить эксплойт до фикса и снова после. Нашлось реально неприятное:

• HTTP response splitting (CWE-113) — Response не проверял заголовки/куки на сырые CR/LF/NUL.

• Pickle RCE в Redis-кэше (CWE-502) — RedisCache.get() вызывал pickle.loads() на произвольных байтах из Redis без аутентификации; всё, что могло записать этот ключ, получало RCE при следующем чтении. Теперь есть secret= для HMAC-подписи значений.

• Cross-site WebSocket hijacking — хендшейк вообще не проверял Origin, так что страница с любого другого сайта могла открыть WebSocket к приложению и прокатиться на cookie-based сессии.

• create_app() по умолчанию поднимался в dev=True — фабрика ASGI, которую документация рекомендует для продакшена (uvicorn endocore.asgi:create_app --factory), включала dev-режим по умолчанию, если переменная окружения не была явно выставлена — тихо открывая /docs, dev-watcher и ослабленную проверку origin.

• Две гонки в ORM (get_or_create/update_or_create и M2M add()) роняли необработанный IntegrityError, когда два вызова конкурировали за одну ещё не существующую строку/связь.

Всё это описано в гайде по безопасности и CHANGELOG. bandit и pip-audit теперь гоняются в CI на каждый push, парсеры запросов (multipart, JSON, query string) property-fuzzed через hypothesis.

После security-аудита я задался вопросом: а что из кода вообще ни разу не выполнялось хоть одним тестом? Пошёл по покрытию — не ради цифры, а потому что почти каждая непокрытая строка оказывалась либо реальным пробелом в поведении, либо забытым краевым случаем. По пути, просто как побочный эффект погони за покрытием (не целенаправленного поиска багов), нашлись три настоящих бага:

• endo test -q -k name был сломан — argparse.parse_known_args() разбивал распознанные и нераспознанные токены на два bucket’а, и при склеивании они теряли относительный порядок, так что -k оставался без значения.

• Manager.ain_bulk() отсутствовал — у каждого другого асинхронного метода QuerySet был делегат на уровне Manager, у этого — нет.

• Race condition в WebSocketManager.start() — метод возвращал управление сразу после запуска фонового потока-подписчика, не дожидаясь, пока Redis реально подтвердит psubscribe(). broadcast() от другого воркера сразу после start() (именно то, что происходит, когда несколько воркеров стартуют примерно одновременно) мог потеряться безвозвратно — Redis pub/sub не повторяет сообщения для опоздавших подписчиков.

Сейчас покрытие — 99.9%+, 2329 тестов, часть из них — против настоящих PostgreSQL и Redis в CI (не только SQLite и фейки). Условие для теста было простое, которое я себе поставил с самого начала: никаких тестов вида assert repr(User()) == "<User>" только ради цифры — каждый тест должен проверять реальное поведение, реальные коды ответов, реальные исключения.

Если вы уже писали на FastAPI — ментальная модель переносится почти без изменений: то же ASGI-развёртывание, похожая форма Request/Response, тот же паттерн Depends(...). Меняется только то, где живёт роут: вместо декоратора там, где кто-то его написал, POST /v1/user/role — это файл Api/v1/User/Role/Post.py.

1.0.0 фиксирует публичный API под semver — в документации есть отдельный раздел API stability, что именно покрыто гарантией, а что (внутренности async_native, точный SQL, шаблоны endo new) может меняться и дальше.

Это личный проект, не корпоративный продукт — делаю его в свободное время, потому что мне было интересно посмотреть, насколько далеко можно довести идею “дерево = API” без потери в безопасности и тестируемости. Буду рад вопросам, придиркам и issue — особенно к архитектурным решениям, которые я, возможно, оправдываю задним числом.

• GitHub: https://github.com/Drakulyonok/endocore

• Документация: https://endocore.readthedocs.io

• Discord: https://discord.gg/jwvGj2M9EX