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:
- Inner
@dataclasss — declaram o surface area por módulo (o que fica visível para os routers) @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:
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:
- Instanciação por request:
AgentRepositoryeCreateAgentsão criados em todo request, mesmo que sejam stateless e possam ser singleton - Invisibilidade: as dependências ficam distribuídas em funções espalhadas — não há um lugar único onde ver "o que esse entrypoint usa"
- 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 |