Pular para conteúdo

Injeção de Dependência

Esta é a decisão mais fácil de "resolver por conta própria" da forma errada. Uma chain de Depends() que parece razoável pode destruir testabilidade e criar acoplamento invisível.


A regra central

Um único Depends(get_container) por rota. Tudo mais vem do container.

# ✅ Correto — um único ponto de entrada
async def create_agent(
    app: AppContainerDep,      # ← um Depends, tudo vem daqui
    tenant_id: TenantDep,
    request: CreateAgentRequest,
) -> AgentDetailResponse:
    command = mapper.to_create_agent_command(request, tenant_id)
    result = await app.agent.create_agent.execute(command)
    return mapper.agent_to_detail_response(result.agent, result.version)

# ❌ Errado — múltiplos Depends() para injetar UCs
async def create_agent(
    create_agent_uc: Annotated[CreateAgent, Depends(get_create_agent_uc)],  # ← não
    agent_repo: Annotated[AgentRepository, Depends(get_agent_repo)],        # ← não
    ...

Se você está criando uma função get_X_uc() para usar em Depends(), está indo na direção errada.


Como o container funciona

Surface containers

Há um container por entrypoint, cada um expõe apenas o que aquele entrypoint precisa:

Container Entrypoint Arquivo
AppContainer App API (porta 8000) src/entrypoints/http/app/compose/container.py
BackofficeContainer Backoffice API (porta 8001) src/entrypoints/http/backoffice/compose/container.py
WorkerContainer Celery Worker src/entrypoints/worker/compose/container.py

Os containers não são compartilhados entre entrypoints. O AppContainer não conhece BackofficeContainer e vice-versa.

Estrutura do AppContainer

O container é um @dataclass com dois tipos de membros:

  1. Inner @dataclasss — declaram o surface area por módulo (o que fica visível para os routers)
  2. @cached_propertys — instanciam dependências de forma lazy (infra + UCs montados)
@dataclass
class AppContainer:
    settings: AppSettings

    # ── Surface area declarations ──────────────────────────────────────
    @dataclass
    class Identity:
        register_user: RegisterUser
        authenticate: Authenticate
        revoke_session: RevokeSession
        user_reader: UserReader
        ...

    @dataclass
    class Agent:
        create_agent: CreateAgent
        update_agent: UpdateAgent
        publish_agent: PublishAgent
        ...

    @dataclass
    class Billing:
        billing_service: BillingService
        subscription_guard: SubscriptionGuard
        ...

    # ── Infrastructure (lazy, nunca tocados se não usados) ─────────────
    @cached_property
    def _engine(self) -> AsyncEngine:
        return create_async_engine(self.settings.database.url, ...)

    @cached_property
    def _session_maker(self) -> async_sessionmaker[AsyncSession]:
        return async_sessionmaker(bind=self._engine, ...)

    @cached_property
    def _uow(self) -> UnitOfWork:
        return PgUnitOfWork(session_maker=self._session_maker)

    @cached_property
    def _redis(self) -> Redis:
        return Redis.from_url(self.settings.redis.url)

    # ── Módulos (lazy, monta tudo quando acessado) ─────────────────────
    @cached_property
    def identity(self) -> Identity:
        repo = IdentityRepository(...)
        return AppContainer.Identity(
            register_user=RegisterUser(repo=repo, uow=self._uow, ...),
            authenticate=Authenticate(repo=repo, jwt=self._jwt_authenticator, ...),
            ...
        )

    @cached_property
    def agent(self) -> Agent:
        ...

Por que @cached_property

@cached_property é lazy: a propriedade só é computada quando acessada pela primeira vez, e o resultado é cacheado no objeto. Se um modulo do container nunca for acessado em um request, seus objetos não são criados.

Isso significa que AppContainer pode ter 200+ UCs e o custo de instanciação de um request que só usa identity é apenas o da propriedade identity e suas dependências diretas.


Singleton via @lru_cache

O container em si é singleton — uma única instância por processo:

@lru_cache(maxsize=1)
def get_app_container() -> AppContainer:
    """Get or create the singleton AppContainer."""
    settings = AppSettings()  # lê variáveis de ambiente
    return AppContainer(settings=settings)

maxsize=1 significa que há exatamente uma entrada no cache, que nunca expira durante a vida do processo. A mesma instância de AppContainer é retornada em todos os requests — e como @cached_property é thread-safe no Python 3.12+, o lazy initialization também é seguro.

Para testes, o cache pode ser limpo:

def reset_app_container_cache() -> None:
    get_app_container.cache_clear()

AppContainerDep e TransactionDep

Em src/entrypoints/http/app/compose/dependencies.py ficam os aliases tipados que os routers usam:

# Alias tipado — um único Depends() por router
AppContainerDep = Annotated[AppContainer, Depends(get_app_container)]

# Transaction via UoW — aberta e fechada pelo container
async def _get_transaction(app: AppContainerDep) -> AsyncGenerator[Transaction, None]:
    async with app.uow.begin() as tx:
        yield tx

TransactionDep = Annotated[Transaction, Depends(_get_transaction)]

TransactionDep resolve dois problemas de uma vez: garante que a transação seja aberta antes do handler e fechada (com commit ou rollback) depois — e que o uow venha sempre do mesmo container singleton, nunca de uma instância avulsa.

Quando usar AppContainerDep vs TransactionDep

Cenário Use
Chamar um UC (escrita) AppContainerDep — o UC abre sua própria transação via uow
Chamar um Query Handler (leitura) TransactionDep — query handlers recebem tx diretamente
Route guard (require_active_subscription) Ambos — o guard usa app para acessar o serviço e tx para o banco

Exemplo real do agents/v1/router.py:

# Escrita → AppContainerDep (UC abre sua própria transação)
async def create_agent(
    tenant_id: TenantDep,
    request: CreateAgentRequest,
    app: AppContainerDep,
) -> AgentDetailResponse:
    command = mapper.to_create_agent_command(request, tenant_id)
    result = await app.agent.create_agent.execute(command)
    return mapper.agent_to_detail_response(result.agent, result.version)

# Leitura → TransactionDep (query handler recebe tx)
async def list_agents(
    tenant_id: TenantDep,
    params: Annotated[ListAgentsParams, Query()],
    tx: TransactionDep,
) -> AgentListResponse:
    query = mapper.to_list_agents_query(params, tenant_id)
    result = await list_agents_handle(query, tx)
    return mapper.agent_list_view_to_response(result)

Por que não usar Depends() diretamente para UCs

O problema com Depends() chains

FastAPI resolve Depends() em grafo. Se você cria:

# ❌ Errado
def get_agent_repo() -> AgentRepository:
    return AgentRepository(engine=get_engine())

def get_create_agent_uc(repo = Depends(get_agent_repo)) -> CreateAgent:
    return CreateAgent(repo=repo, uow=get_uow())

Você tem três problemas:

  1. Instanciação por request: AgentRepository e CreateAgent são criados em todo request, mesmo que sejam stateless e possam ser singleton
  2. Invisibilidade: as dependências ficam distribuídas em funções espalhadas — não há um lugar único onde ver "o que esse entrypoint usa"
  3. Testabilidade: para testar um router, você precisa fazer override de N Depends() individuais em vez de substituir um único container

O container como contrato explícito

Os inner @dataclasss do container são um contrato explícito do que cada surface expõe:

@dataclass
class Agent:
    create_agent: CreateAgent
    update_agent: UpdateAgent
    publish_agent: PublishAgent
    archive_agent: ArchiveAgent
    ...

Se um UC não está declarado aqui, os routers não têm como acessá-lo. Isso torna a surface area auditável: para saber o que o App API pode fazer, leia os inner dataclasses do AppContainer.


Erros comuns e como identificá-los

❌ Instanciar UC diretamente no router

# Errado
async def create_agent(request: CreateAgentRequest, ...):
    repo = AgentRepository(engine=engine)    # ← viola DI
    uc = CreateAgent(repo=repo, uow=uow)     # ← não vem do container
    result = await uc.execute(command)

Correção: app.agent.create_agent via AppContainerDep.


❌ Criar função get_X() para usar com Depends()

# Errado
def get_create_agent_uc() -> CreateAgent:
    container = get_app_container()
    return container.agent.create_agent  # ← inútil, já tem AppContainerDep

async def create_agent(uc: Annotated[CreateAgent, Depends(get_create_agent_uc)]):
    ...

Correção: injete AppContainerDep e use app.agent.create_agent diretamente. Funções get_X_uc() são indireção desnecessária.


❌ Passar AppContainer para dentro de um UC ou Service

# Errado — UC não conhece o container
class CreateAgent:
    def __init__(self, container: AppContainer):  # ← viola Clean Architecture
        self.repo = container._agent_repo
        self.uow = container._uow

Regra: UCs recebem dependências concretas (repo, uow, service) — nunca o container. O container existe apenas na camada de entrypoints (HTTP layer + worker).


❌ Acessar get_app_container() dentro de um Service ou Repository

# Errado
class AgentService:
    def process(self):
        container = get_app_container()    # ← Service não acessa container
        repo = container._agent_repo

get_app_container() só deve ser chamado em: - dependencies.py (via AppContainerDep) - Rate limiters e guards em dependencies.py - Lifespan handlers (startup / shutdown) - Testes (via reset_app_container_cache)


❌ Criar um segundo container para o mesmo entrypoint

# Errado — dois containers para o mesmo entrypoint
class LightweightContainer:
    def __init__(self):
        self.repo = AgentRepository()
        ...

Regra: um container por surface. Se o AppContainer parece grande demais, a solução é melhorar a organização dos inner dataclasses, não criar um segundo container.


Referência rápida para code review

O que ver no código O que significa
Depends(get_X_uc) onde get_X_uc retorna um UC do container Indireção desnecessária — usar AppContainerDep diretamente
AgentRepository() ou CreateAgent() instanciados em router ou UC Violação de DI — deve vir do container
get_app_container() dentro de Service, Repository ou UC Container só pertence à camada de entrypoints
Novo arquivo get_X_container() além dos três existentes Duplicação de container — adicionar ao container existente
AppContainer passado como parâmetro para um UC UC deve receber dependências, não o container