Dependency Injection Architecture¶

# ✅ CORRECT: Return protocol type, create implementation
def get_user_repository(session: AsyncSession) -> UserRepository:
    from src.infrastructure.persistence.repositories.user_repository import (
        UserRepository as UserRepositoryImpl
    )
    return UserRepositoryImpl(session=session)

class RegisterUserHandler:
    def __init__(
        self,
        user_repo: UserRepository,           # Protocol from domain
        password_hasher: PasswordHashingProtocol,  # Protocol from domain
        event_bus: EventBusProtocol,         # Protocol from domain
    ):
        self._user_repo = user_repo
        self._password_hasher = password_hasher
        self._event_bus = event_bus

src/core/container/
├── __init__.py              # Re-exports all (single entry point preserved)
├── infrastructure.py        # Core services: cache, db, secrets, logging, etc.
├── events.py               # Event bus with all 27 subscriptions
├── repositories.py         # All repository factories
├── handler_factory.py      # Auto-wired CQRS handler injection (38 handlers)
├── providers.py            # Financial provider adapter factory
└── authorization.py        # Casbin RBAC enforcer

# Single import point preserved
from src.core.container import get_cache, get_db_session, get_register_user_handler

# Or import from specific module for clarity
from src.core.container.infrastructure import get_cache
from src.core.container.auth_handlers import get_register_user_handler

# src/domain/entities/user.py
# Domain entities are pure - NO container imports
@dataclass(slots=True, kw_only=True)
class User:
    id: UUID
    email: str
    # Pure business logic only

# src/application/commands/handlers/register_user_handler.py
# NO container imports - only domain protocols!
from src.domain.protocols import UserRepository, PasswordHashingProtocol
from src.domain.protocols.event_bus_protocol import EventBusProtocol

class RegisterUserHandler:
    """Handler receives all dependencies via constructor."""

    def __init__(
        self,
        user_repo: UserRepository,              # ← Injected protocol
        password_service: PasswordHashingProtocol,  # ← Injected protocol
        event_bus: EventBusProtocol,            # ← Injected protocol
    ) -> None:
        self._user_repo = user_repo
        self._password_service = password_service
        self._event_bus = event_bus

    async def handle(self, cmd: RegisterUser) -> Result[UUID, str]:
        # Use injected dependencies - no container knowledge
        password_hash = self._password_service.hash_password(cmd.password)
        user = User(email=cmd.email, password_hash=password_hash, ...)
        await self._user_repo.save(user)
        await self._event_bus.publish(UserRegistrationSucceeded(...))
        return Success(value=user.id)

async def get_register_user_handler(
    session: AsyncSession = Depends(get_db_session),
) -> RegisterUserHandler:
    """Container factory wires dependencies."""
    from src.infrastructure.persistence.repositories import UserRepository

    return RegisterUserHandler(
        user_repo=UserRepository(session=session),
        password_service=get_password_service(),  # App-scoped singleton
        event_bus=get_event_bus(),                # App-scoped singleton
    )

# src/infrastructure/providers/schwab_provider.py
from src.core.container import get_secrets

class SchwabProvider:
    def __init__(self):
        # Infrastructure can use container
        secrets = get_secrets()
        self.api_key = secrets.get_secret("schwab/api_key")

# src/presentation/api/v1/users.py
from fastapi import Depends

@router.post("/users", status_code=201)
async def create_user(
    data: UserCreate,
    session: AsyncSession = Depends(get_db_session),  # ← Request-scoped
    cache: CacheProtocol = Depends(get_cache),        # ← App-scoped
    handler: RegisterUserHandler = Depends(get_register_user_handler),  # ← Request-scoped
) -> UserResponse:
    """Create new user.

    FastAPI automatically injects dependencies:
        - get_db_session(): New session per request
        - get_cache(): Shared cache singleton
        - get_register_user_handler(): New handler per request
    """
    result = await handler.handle(
        RegisterUser(email=data.email, password=data.password),
        session
    )

    match result:
        case Success(user_id):
            return UserResponse(id=user_id, email=data.email)
        case Failure(error):
            raise HTTPException(400, detail=error.message)

# tests/unit/test_application_register_user_handler.py
from unittest.mock import AsyncMock, Mock
import pytest

@pytest.mark.asyncio
async def test_register_user_handler_success():
    """Test handler with mock dependencies - NO patching needed."""
    # Create mock dependencies directly
    mock_user_repo = Mock()
    mock_user_repo.find_by_email = AsyncMock(return_value=None)  # No existing user
    mock_user_repo.save = AsyncMock()

    mock_password_service = Mock()
    mock_password_service.hash_password.return_value = "hashed_password"

    mock_event_bus = Mock()
    mock_event_bus.publish = AsyncMock()

    # Instantiate handler with mocks
    handler = RegisterUserHandler(
        user_repo=mock_user_repo,
        password_service=mock_password_service,
        event_bus=mock_event_bus,
    )

    # Test the handler
    result = await handler.handle(RegisterUser(email="test@example.com", password="secure"))

    assert isinstance(result, Success)
    mock_user_repo.save.assert_called_once()
    mock_event_bus.publish.assert_called()  # Events emitted

# tests/integration/test_cache_redis.py
from src.core.container import get_cache

async def test_cache_integration():
    """Test with real Redis (integration test)."""
    cache = get_cache()  # Real Redis adapter

    result = await cache.set("test:key", "value")
    assert result is Success

    result = await cache.get("test:key")
    assert result.value == "value"

# tests/api/test_users_endpoints.py
from fastapi.testclient import TestClient

def test_create_user_endpoint(client: TestClient):
    """Test endpoint with real dependencies."""
    # TestClient uses real container (test environment)
    response = client.post("/api/v1/users", json={
        "email": "test@example.com",
        "password": "SecurePass123!"
    })

    assert response.status_code == 201

# src/main.py
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan - initialize singletons."""
    # Singletons created on first access (lazy initialization)
    get_cache()      # ← Creates Redis connection pool
    get_secrets()    # ← Creates secrets adapter
    get_database()   # ← Creates database connection pool

    yield

    # Cleanup
    await close_cache()
    await close_database()

app = FastAPI(lifespan=lifespan)

# src/infrastructure/cache/__init__.py
def get_cache():  # ← Singleton in __init__.py
    ...

# src/infrastructure/secrets/factory.py
def create_secrets_manager():  # ← Factory function
    ...

# Direct instantiation
db = Database(settings.database_url)  # ← No abstraction

# src/core/container.py
@lru_cache()
def get_cache(): ...

@lru_cache()
def get_secrets(): ...

@lru_cache()
def get_database(): ...

# Old: Import from infrastructure modules
from src.infrastructure.cache import get_cache  # Singleton in __init__.py
from src.infrastructure.secrets.env_adapter import EnvAdapter  # Direct import
from src.infrastructure.persistence.database import Database  # Direct instantiation

# New: All dependencies from container
from src.core.container import get_cache, get_secrets, get_database

# WRONG - Don't create dependencies directly
class RegisterUserHandler:
    def __init__(self):
        self.cache = RedisAdapter(...)  # ❌ Tight coupling!

# WRONG - Domain should be pure
# src/domain/entities/user.py
from src.core.container import get_cache  # ❌ Domain depends on infrastructure!

# WRONG - Don't use singleton for request data
@lru_cache()
def get_current_user():  # ❌ Current user should be request-scoped!
    ...

# WRONG - Don't create new connections per request
async def get_cache():  # ❌ No @lru_cache - creates new Redis connection per request!
    return RedisAdapter(...)

# Command handler - depends only on protocols
class RegisterUserHandler:
    def __init__(self, user_repo: UserRepository, event_bus: EventBusProtocol):
        self._user_repo = user_repo
        self._event_bus = event_bus

# Query handler with ownership verification - uses OwnershipVerifier service
class GetAccountHandler:
    def __init__(self, ownership_verifier: OwnershipVerifier):
        self._verifier = ownership_verifier

    async def handle(self, query: GetAccount) -> Result[AccountResult, str]:
        result = await self._verifier.verify_account_ownership(
            query.account_id, query.user_id
        )
        if isinstance(result, Failure):
            return Failure(error=error_map[result.error.code])
        account = result.value
        # ... map to DTO

# Presentation layer uses handler_factory for auto-wiring
from src.core.container.handler_factory import handler_factory

@router.get("/accounts/{account_id}")
async def get_account(
    handler: GetAccountHandler = Depends(handler_factory(GetAccountHandler))
):
    result = await handler.handle(query)
    ...

# src/core/container/handler_factory.py
SESSION_SERVICE_TYPES: set[type] = {
    OwnershipVerifier,  # Needs repos for ownership chain verification
}

def _get_session_service_instance(service_type: type, session: AsyncSession) -> Any:
    """Create session-scoped service instance."""
    if service_type is OwnershipVerifier:
        return OwnershipVerifier(
            transaction_repo=TransactionRepository(session=session),
            holding_repo=HoldingRepository(session=session),
            account_repo=AccountRepository(session=session),
            connection_repo=ProviderConnectionRepository(session=session),
        )
    raise ValueError(f"Unknown session service type: {service_type}")

# Repositories receive session from presentation layer
class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session  # ← Injected, not created

# Presentation layer provides session
@router.get("/users/{id}")
async def get_user(
    id: UUID,
    session: AsyncSession = Depends(get_db_session)
):
    repo = UserRepository(session)
    user = await repo.find_by_id(id)
    ...

┌───────────────┐                     ┌───────────────┐
│      API      │                     │  dashtam-jobs │
│  JobsMonitor  │───── same Redis ────│    TaskIQ     │
│               │      queue name     │               │
└───────────────┘                     └───────────────┘
        │                                     │
        │  LLEN dashtam:jobs                 │ LPUSH/BRPOP dashtam:jobs
        │  PING                              │ Store results
        └─────────────┬───────────────────────┘
                      ▼
              ┌───────────────┐
              │     Redis     │
              │  (shared)     │
              └───────────────┘

# .env.dev
REDIS_URL=redis://redis:6379/0
# JOBS_REDIS_URL not set - falls back to REDIS_URL
JOBS_QUEUE_NAME=dashtam:jobs

# docker-compose.yml - add second Redis service
redis-jobs:
  image: redis:8.4-alpine
  container_name: dashtam-redis-jobs
  volumes:
    - redis_jobs_data:/data
  command: redis-server --appendonly yes
  networks:
    - dashtam-network

# .env.prod
REDIS_URL=redis://redis:6379/0           # Cache
JOBS_REDIS_URL=redis://redis-jobs:6379/0  # Jobs queue
JOBS_QUEUE_NAME=dashtam:jobs

# Application layer - direct use
from src.core.container import get_jobs_monitor

monitor = get_jobs_monitor()
health = await monitor.check_health()
if health.value.healthy:
    print(f"Queue length: {health.value.queue_length}")

# Presentation layer - FastAPI Depends
from fastapi import Depends
from src.core.container import get_jobs_monitor

@router.get("/admin/jobs")
async def get_jobs_status(
    monitor: JobsMonitor = Depends(get_jobs_monitor)
):
    result = await monitor.check_health()
    ...