260615 Python DI IoC 라이브러리 리서치

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


def get_repository() -> "UserRepository":
    return UserRepository()


@app.get("/users/{user_id}")
def get_user(
    user_id: int,
    repo: Annotated["UserRepository", Depends(get_repository)],
):
    return repo.get(user_id)

from dataclasses import dataclass
from typing import Protocol

from dependency_injector import containers, providers


class UserRepository(Protocol):
    def get_name(self, user_id: int) -> str:
        ...


class InMemoryUserRepository:
    def __init__(self, users: dict[int, str]) -> None:
        self._users = users

    def get_name(self, user_id: int) -> str:
        return self._users[user_id]


@dataclass
class UserService:
    repository: UserRepository

    def greeting(self, user_id: int) -> str:
        return f"Hello, {self.repository.get_name(user_id)}"


class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    user_repository = providers.Singleton(
        InMemoryUserRepository,
        users=config.users,
    )

    user_service = providers.Factory(
        UserService,
        repository=user_repository,
    )


def main() -> None:
    container = Container()
    container.config.users.from_value({1: "Alice", 2: "Bob"})

    service = container.user_service()
    print(service.greeting(1))


if __name__ == "__main__":
    main()

from unittest import mock


def test_user_service_greeting() -> None:
    container = Container()
    repository = mock.Mock()
    repository.get_name.return_value = "TestUser"

    with container.user_repository.override(repository):
        service = container.user_service()
        assert service.greeting(10) == "Hello, TestUser"

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="APP_")

    database_url: str = "sqlite:///./app.db"
    external_api_base_url: str = "https://api.example.com"
    external_api_timeout: float = 3.0

from dataclasses import dataclass
from typing import Protocol


class UserRepository(Protocol):
    def get_name(self, user_id: int) -> str:
        ...


class AuditClient(Protocol):
    def write_event(self, event_name: str, payload: dict) -> None:
        ...


@dataclass
class UserService:
    repository: UserRepository
    audit_client: AuditClient

    def get_profile_message(self, user_id: int) -> str:
        name = self.repository.get_name(user_id)
        self.audit_client.write_event("profile_read", {"user_id": user_id})
        return f"User profile: {name}"

class SqlUserRepository:
    def __init__(self, session_factory):
        self._session_factory = session_factory

    def get_name(self, user_id: int) -> str:
        with self._session_factory() as session:
            row = session.execute(
                "select name from users where id = :id",
                {"id": user_id},
            ).one()
            return row[0]

import httpx


class HttpAuditClient:
    def __init__(self, base_url: str, timeout: float) -> None:
        self._client = httpx.Client(base_url=base_url, timeout=timeout)

    def write_event(self, event_name: str, payload: dict) -> None:
        self._client.post("/audit", json={"event": event_name, "payload": payload})

    def close(self) -> None:
        self._client.close()

from contextlib import contextmanager

from dependency_injector import containers, providers
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from .clients import HttpAuditClient
from .repositories import SqlUserRepository
from .services import UserService


@contextmanager
def init_session_factory(database_url: str):
    engine = create_engine(database_url)
    session_factory = sessionmaker(bind=engine)
    try:
        yield session_factory
    finally:
        engine.dispose()


@contextmanager
def init_audit_client(base_url: str, timeout: float):
    client = HttpAuditClient(base_url=base_url, timeout=timeout)
    try:
        yield client
    finally:
        client.close()


class Container(containers.DeclarativeContainer):
    wiring_config = containers.WiringConfiguration(modules=[".api"])

    config = providers.Configuration()

    session_factory = providers.Resource(
        init_session_factory,
        database_url=config.database_url,
    )

    audit_client = providers.Resource(
        init_audit_client,
        base_url=config.external_api_base_url,
        timeout=config.external_api_timeout,
    )

    user_repository = providers.Factory(
        SqlUserRepository,
        session_factory=session_factory,
    )

    user_service = providers.Factory(
        UserService,
        repository=user_repository,
        audit_client=audit_client,
    )

from typing import Annotated

from dependency_injector.wiring import Provide, inject
from fastapi import APIRouter, Depends

from .containers import Container
from .services import UserService

router = APIRouter()


@router.get("/users/{user_id}")
@inject
def get_user(
    user_id: int,
    service: Annotated[UserService, Depends(Provide[Container.user_service])],
) -> dict[str, str]:
    return {"message": service.get_profile_message(user_id)}

from fastapi import FastAPI

from .api import router
from .containers import Container
from .settings import Settings


def create_app() -> FastAPI:
    container = Container()
    container.config.from_pydantic(Settings())
    container.init_resources()

    app = FastAPI()
    app.container = container
    app.include_router(router)

    @app.on_event("shutdown")
    def shutdown() -> None:
        container.shutdown_resources()

    return app


app = create_app()