adc-appkit

ADC AppKit

Компонентная архитектура для async Python приложений: декларативные компоненты, dependency injection, request scope.

Install

pip install adc-appkit

Quick start

import asyncio
from adc_appkit import BaseApp, component
from adc_appkit.components.component import Component

class Database:
    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port

    async def close(self):
        pass

class DatabaseComponent(Component[Database]):
    async def _start(self, host: str, port: int, **kw) -> Database:
        return Database(host, port)

    async def _stop(self) -> None:
        await self.obj.close()

class App(BaseApp):
    db = component(DatabaseComponent, config_key="db")

async def main():
    app = App(components_config={"db": {"host": "localhost", "port": 5432}})
    await app.start()

    # дескриптор возвращает объект напрямую
    print(app.db.host)  # "localhost"

    health = await app.healthcheck()  # {"db": True}
    await app.stop()

asyncio.run(main())

Core concepts

Component[T]

Базовый класс для компонентов с управляемым жизненным циклом:

class PGComponent(Component[Pool]):
    async def _start(self, dsn: str, **kw) -> Pool:
        return await asyncpg.create_pool(dsn)

    async def _stop(self) -> None:
        await self.obj.close()

    async def is_alive(self) -> bool:
        return not self.obj._closed

• _start(**config) — создает и возвращает управляемый объект
• _stop() — освобождает ресурсы
• is_alive() — healthcheck (по умолчанию True)

create_component(cls)

Оборачивает обычный класс в Component без написания boilerplate. Конструктор класса вызывается с параметрами из конфига:

from adc_appkit.components.component import create_component

class DAO:
    def __init__(self, pool: Pool):
        self.pool = pool

    async def get_user(self, user_id):
        return await self.pool.fetchrow("SELECT * FROM users WHERE id=$1", user_id)

class App(BaseApp):
    pg = component(PGComponent, config_key="pg")
    dao = component(create_component(DAO), config_key="dao", dependencies={"pool": "pg"})

При остановке create_component автоматически вызывает close() (sync или async) на объекте, если метод существует.

Strategies: SINGLETON vs REQUEST

from adc_appkit import ComponentStrategy

class App(BaseApp):
    # создается один раз при app.start(), живет до app.stop()
    pg = component(PGComponent, config_key="pg")

    # создается на каждый request_scope, уничтожается при выходе
    current_user = component(
        CurrentUserComponent,
        config_key="current_user",
        strategy=ComponentStrategy.REQUEST,
    )

SINGLETON (по умолчанию) — один экземпляр на всё приложение. REQUEST — новый экземпляр на каждый request scope; изолирован через contextvars.ContextVar.

Dependency injection

Зависимости объявляются как маппинг {имя_параметра: имя_компонента}. При старте компонента в _start() автоматически передается .obj зависимости:

class App(BaseApp):
    pg = component(PGComponent, config_key="pg")
    dao = component(create_component(DAO), config_key="dao", dependencies={"pool": "pg"})
    #                                                                       ^^^^^   ^^^^
    #                                                        параметр в __init__    имя компонента

Правила:

• SINGLETON может зависеть от SINGLETON
• REQUEST может зависеть от SINGLETON и REQUEST
• SINGLETON не может зависеть от REQUEST (защита от утечки request-scoped состояния)

Компоненты запускаются в топологическом порядке: зависимости стартуют раньше зависимых.

Request scope

Request scope управляет жизненным циклом REQUEST компонентов. Контекст (ctx) передает параметры, специфичные для запроса:

async with app.request_scope({"current_user": {"user_id": uid}}) as scope:
    # все REQUEST компоненты созданы и запущены
    user = app.current_user  # obj напрямую через дескриптор
    # или
    user = scope.use("current_user")

Конфиг REQUEST компонента собирается в порядке: base config -> ctx overrides -> DI injection. Ключ в ctx должен совпадать с config_key компонента.

Real-world patterns

App с PG + DAO + request-scoped identity

Паттерн из реального auth-сервиса: PG pool (SINGLETON) -> DAO (SINGLETON, зависит от PG) -> CurrentIdentity (REQUEST, получает sub из JWT и dao как зависимость):

from adc_appkit import BaseApp, ComponentStrategy, component
from adc_appkit.components.component import Component, create_component
from adc_appkit.components.pg import PG

class DAO:
    def __init__(self, pool):
        self.pool = pool

    async def get_user_by_id(self, user_id):
        return await self.pool.fetchrow("SELECT * FROM users WHERE id=$1", user_id)

class CurrentIdentity(Component):
    """REQUEST-scoped: загружает текущего пользователя по ID из JWT."""

    async def _start(self, sub, dao, **kw):
        user = await dao.get_user_by_id(sub)
        if not user:
            raise ValueError(f"User {sub} not found")
        return user

    async def _stop(self):
        pass

class App(BaseApp):
    pg = component(PG, config_key="pg")
    dao = component(create_component(DAO), config_key="dao", dependencies={"pool": "pg"})
    current_identity = component(
        CurrentIdentity,
        config_key="current_identity",
        strategy=ComponentStrategy.REQUEST,
        dependencies={"dao": "dao"},
    )

Endpoint: request_scope с контекстом из JWT

async def logout_handler(request):
    app: App = request.app.state.app
    user_id = decode_jwt(request.headers["Authorization"]).sub

    async with app.request_scope({"current_identity": {"sub": user_id}}):
        # app.current_identity — загруженный из БД пользователь
        await app.revoke_session(session_id)

Значение "sub" из ctx объединяется с base config и DI-зависимостью dao, а затем передается в CurrentIdentity._start(sub=..., dao=...).

Healthcheck

health = await app.healthcheck()
# {"pg": True, "dao": True}  — только SINGLETON компоненты

Architecture

• Топологическая сортировка — компоненты запускаются и останавливаются в порядке, учитывающем зависимости
• ContextVar изоляция — каждый request scope хранит свой кэш компонентов в contextvars.ContextVar, безопасно для concurrent asyncio tasks
• Сборка через MRO — дескрипторы компонентов собираются с учетом наследования (reversed(cls.mro()))
• Graceful cleanup — при ошибке старта REQUEST компонента уже запущенные компоненты корректно останавливаются

Built-in components

Development

uv sync --dev
uv run pytest -v
uv run black adc_appkit tests
uv run isort adc_appkit tests
uv run mypy adc_appkit