Search
🌍

DI는 “소스코드 입장”에서 논리적으로 모든 의존성이 컨테이너로 향하도록 하고, 컨테이너에서 의존성을 중앙화된 방식으로 관리할 수 있다는 장점 때문에 사용한다.

🚀 prev note
♻️ prev note
16 more properties
클린 아키텍처에서는 애플리케이션 계층(Application Layer)이 인프라스트럭처 계층(Infrastructure Layer)에 의존해서는 안된다. 대신 추상화(인터페이스)에 의존해야 하고, 구체적인 구현체는 실행 시점에 주입되어야 한다.
# 도메인 계층 (추상화) from abc import ABC, abstractmethod class UserRepositoryInterface(ABC): @abstractmethod def get_by_email(self, email: str) -> Optional[User]: pass @abstractmethod def create(self, user: User) -> User: pass # 애플리케이션 계층 class UserService: def __init__(self, user_repo: UserRepositoryInterface): # 추상화에 의존 self.user_repo = user_repo def authenticate_user(self, email: str, password: str): user = self.user_repo.get_by_email(email) # 인증 로직... # 인프라스트럭처 계층 class UserRepository(UserRepositoryInterface): # 구체적인 구현 def __init__(self, db: Session): self.db = db def get_by_email(self, email: str) -> Optional[User]: return self.db.query(User).filter(User.email == email).first() # 표현 계층 @router.post("/login/") def login( credentials: LoginCredentials, user_service: UserService, ): return user_service.authenticate_user(credentials.email, credentials.password)
Python
복사
하지만 FastAPI의 Depends()만으로는 이런 구조를 자연스럽게 구현하기 어렵다. user_repo: UserRepositoryInterface = Depends(SupabaseUserRepository) 라고 작성하는 순간, UserService가 SupabaseUserRepository에 의존해 버리는 모순이 발생한다.
class UserRepositoryInterface(ABC): ... # 애플리케이션 계층 class UserService: def __init__( self, user_repo: UserRepositoryInterface = Depends(UserRepository) ): ... # 인프라스트럭처 계층 def get_db(): db = SessionLocal() try: yield db db.commit() except: db.rollback() raise finally: db.close() class UserRepository(UserRepositoryInterface): def __init__( self, db: Session = Depends(get_db) ): ... # 표현 계층 def get_service(): return UserService() @router.post("/login/") def login( credentials: LoginCredentials, user_service: UserService = Depends(get_service), ): ...
Python
복사
이 문제를 해결하려면 IoC(Inversion of Control) 컨테이너를 사용해야 한다(ref1). Python에서는 dependency-injector 라이브러리가 이런 목적으로 널리 사용된다.
pip install dependency-injector
Shell
복사
IoC 컨테이너를 설정한다.
# app/container.py from dependency_injector import containers, providers from dependency_injector.wiring import Provide, inject from app.db.session import get_db from app.repositories.user_repository import SupabaseUserRepository from app.services.user_service import UserService class Container(containers.DeclarativeContainer): # 설정 config = providers.Singleton(Config) # 데이터베이스 세션 (FastAPI에서 자주 등장하는 get_db 활용) db_session = providers.Resource(get_db) # 리포지토리 계층 user_repository = providers.Factory( SupabaseUserRepository, db=db_session ) # 서비스 계층 user_service = providers.Factory( UserService, user_repo=user_repository )
Python
복사
dependency-injector는 여러 종류의 프로바이더를 제공한다.
•
Factory: 매번 새로운 인스턴스를 생성함. 상태를 가지지 않는 서비스나 요청별로 독립적이어야 하는 객체에 적합
•
Singleton: 첫 번째 호출 시에만 인스턴스를 생성하고, 이후에는 같은 인스턴스를 재사용함. 설정 객체나 캐시 같은 전역적으로 공유되어야 하는 객체에 적합
•
Resource: 생성과 소멸을 관리하는 리소스(데이터베이스 연결, 파일 핸들 등)에 사용
먼저 FastAPI 엔드포인트에서 구현체 대신 IoC 컨테이너를 연결한다.
# app/api/endpoints/users.py from dependency_injector.wiring import Provide, inject from app.container import Container from app.services.user_service import UserService @router.post("/login/") @inject def login( credentials: LoginCredentials, user_service: UserService = Depends(Provide[Container.user_service]), ): return user_service.authenticate_user(credentials.email, credentials.password)
Python
복사
변화된 부분인 Depends(Provide[Container.user_service])에 주목해 보자. 이는 FastAPI의 Depends()와 dependency-injector의 Provide를 결합한 것이다. Provide[Container.user_service]는 IoC 컨테이너에서 UserService 인스턴스를 가져오는 함수를 생성하고, Depends()는 이 함수를 FastAPI의 의존성 주입 시스템에 등록한다.
Before:
---
title: Before - FastAPI Depends()만 사용 (의존성 역전 없음)
---
classDiagram
    class LoginController {
        +login(credentials)
    }
    
    class UserService {
        +authenticate_user()
        -user_repo: UserRepository
    }
    
    class UserRepository {
        +get_by_email()
        +create()
        -db: Session
    }
    
    LoginController --> UserService : Depends(UserService)
    UserService --> UserRepository : Depends(UserRepository)
    
    note for LoginController "컨트롤러가 구체 클래스에 직접 의존"
Mermaid
복사
After:
---
title: After - IoC 컨테이너 사용, 소스코드 PoV
---
classDiagram
    class LoginController {
        +login(credentials)
    }
        
    class UserService {
        +authenticate_user()
        -user_repo: UserRepositoryInterface
    }
        
    class SupabaseUserRepository {
        +get_by_email()
        +create()
        -db: Session
    }
    
    class Container {
        +user_service: Factory
        +user_repository: Factory
    }
    
    LoginController --> Container : Depends(Provide[Container.user_service])
    UserService --> Container : Depends(Provide[Container.user_repository])
    SupabaseUserRepository --> Container : Depends(Provide[Container.db_session])
Mermaid
복사
---
title: After - IoC 컨테이너 사용, 컨테이너 내부 PoV
---
classDiagram
    class LoginController {
        +login(credentials)
    }
        
    class UserService {
        +authenticate_user()
        -user_repo: UserRepositoryInterface
    }
        
    class SupabaseUserRepository {
        +get_by_email()
        +create()
        -db: Session
    }

    class Container {
        +user_service: Factory
        +user_repository: Factory
    }
    
    Container ..> UserService : creates
    Container ..> SupabaseUserRepository : creates
 
    LoginController --> UserService
    UserService --> SupabaseUserRepository
Mermaid
복사
소스코드 입장에서 논리적으로는 컨테이너에 의존하는 모양새가 된다. 그 덕분에 컨테이너 내부에서는 마치 조물주가 피조물들에 영혼을 불어넣는 과정을 제어하듯, 어떤 구현체를 사용할지를 선택할 수 있어 소스코드의 강력한 중앙 관리자의 역할을 할 수 있게 된다.
마지막으로 FastAPI 앱을 초기화할 때 컨테이너를 설정한다.
# app/main.py from fastapi import FastAPI from app.container import Container from app.api.endpoints import users def create_app() -> FastAPI: container = Container() container.config.database_url.from_env("DATABASE_URL") app = FastAPI() app.container = container # 의존성 주입 와이어링 container.wire(modules=[users]) app.include_router(users.router) return app app = create_app()
Python
복사
IoC 컨테이너를 사용하는 경우, 다음과 같이 간단하게 테스트용 repository로 교체할 수 있다.
# tests/conftest.py @pytest.fixture def container(): container = Container() # 필요시 특정 의존성만 Mock으로 교체 container.user_repository.override( providers.Factory(MockUserRepository) ) return container @pytest.fixture def client(container): app = create_app() app.container = container container.wire(modules=[users]) with TestClient(app) as c: yield c
Python
복사
이렇게 FastAPI로 클린 아키텍처의 원칙을 준수하는 애플리케이션을 구축할 수 있다.
하지만 IoC 컨테이너와 데코레이터를 함께 사용할 때에는 하나 더 고려해야 할 점이 있다. 클린 아키텍처에서 서비스 레이어와 리포지토리 레이어를 분리하면, FastAPI의 Depends(get_db)가 제공하는 자동 세션 관리 기능을 직접적으로 활용하기 어렵다.
이 문제를 해결하는 가장 간단한 방법은 데코레이터가 세션을 직접 생성하고 관리하는 것이었다.
from functools import wraps from app.db.session import SessionLocal def with_db_session(func): @wraps(func) def wrapper(self, *args, **kwargs): db = SessionLocal() try: result = func(self, db, *args, **kwargs) db.commit() return result except Exception: db.rollback() raise finally: db.close() return wrapper class SupabaseUserRepository(UserRepositoryInterface): @with_db_session def create_user(self, db: Session, user_data: UserCreate) -> User: user = User(**user_data.dict()) db.add(user) return user @with_db_session def get_by_email(self, db: Session, email: str) -> Optional[User]: return db.query(User).filter(User.email == email).first()
Python
복사
하지만 이 방식은 데코레이터가 데이터베이스 세션 생성 방식을 포함한다는 문제가 있다. 세션 생성방식 관리가 IoC 컨테이너 외부에서 이루어져야 하므로, 테스트 시 다른 DB(e.g. sqlite)나 다른 테이블을 사용할 필요가 있는 경우 변경 작업이 까다롭다. 이들을 제어하는 작업도 DI 컨테이너에 포함되어야 한다.
---
title: After - Session과 Engine도 Container에서 관리
---
classDiagram
    class LoginController {
        +login(credentials)
    }
        
    class UserService {
        +authenticate_user()
        -user_repo: UserRepositoryInterface
    }
        
    class SupabaseUserRepository {
        +get_by_email()
        +create()
        -db: Session
    }

    class SqlAlchemySession {
		    +engine: Engine
    }

    class SqlAlchemyEngine {
		    +database_url: String
    }

    class Container {
        +user_service: Factory
        +user_repository: Factory
    }
    
    Container ..> UserService : creates
    Container ..> SupabaseUserRepository : creates
    Container ..> SqlAlchemySession : creates
    Container ..> SqlAlchemyEngine : creates 
    
    LoginController --> UserService
    UserService --> SupabaseUserRepository
    SupabaseUserRepository --> SqlAlchemySession
    SqlAlchemySession --> SqlAlchemyEngine
Mermaid
복사
IoC 컨테이너를 통해 세션 팩토리를 주입하면 이 문제가 해결된다.
# app/decorators.py from functools import wraps from typing import Callable def with_db_session(session_factory: Callable): def decorator(func): @wraps(func) def wrapper(self, *args, **kwargs): db = session_factory() try: result = func(self, db, *args, **kwargs) db.commit() return result except Exception: db.rollback() raise finally: db.close() return wrapper return decorator # app/container.py from dependency_injector import containers, providers class Container(containers.DeclarativeContainer): config = providers.Configuration() # 데이터베이스 관련 db_engine = providers.Singleton( create_engine, config.database_url, echo=config.db_echo.as_(bool) ) from sqlalchemy.orm import sessionmaker db_session_factory = providers.Factory( sessionmaker, bind=db_engine, autocommit=False, autoflush=False ) # 데코레이터 팩토리 db_session_decorator = providers.Factory( with_db_session, session_factory=db_session_factory ) # 리포지토리 (데코레이터 적용된 버전) user_repository = providers.Factory( SupabaseUserRepository, session_decorator=db_session_decorator ) # Repository에서는 메서드들에 데코레이터 적용하는 방식으로 사용 class SupabaseUserRepository(UserRepositoryInterface): def __init__(self, decorator): # 데코레이터를 받아서 메서드들에 적용 self.with_db_session = decorator self.create_user = self.with_db_session(self._create_user) self.get_by_email = self.with_db_session(self._get_by_email)
Python
복사
데코레이터를 session_factory를 주입받는 데코레이터로 한번 더 감싸고, session_factory는 IoC 컨테이너에서 주입받는 방식이다.
현재 구현은 쓰레드 안전할까?
•
Engine: thread-safe, 여러 스레드에서 공유 가능
•
sessionmaker: thread-safe, 여러 스레드에서 공유 가능
•
Session 객체: thread-local이어야 함, 공유하면 안됨, 하지만 이미 현재 구현은 각 호출마다 db = session_factory()을 이용해 새 세션을 생성하기 때문에 thread-safe!
그리고 리포지토리에서는 주입받은 데코레이터를 사용한다.
class UserRepository(UserRepositoryInterface): def __init__(self, session_decorator): self.with_session = session_decorator # 메서드에 데코레이터 동적 적용 self.create_user = self.with_session(self._create_user) self.get_by_email = self.with_session(self._get_by_email) def _create_user(self, db: Session, user_data: UserCreate) -> User: user = User(**user_data.dict()) db.add(user) return user def _get_by_email(self, db: Session, email: str) -> Optional[User]: return db.query(User).filter(User.email == email).first()
Python
복사
컨텍스트 매니저를 사용하는 것도 방법이다.
from contextlib import contextmanager from typing import Generator class DatabaseContext: def __init__(self, session_factory): self.session_factory = session_factory @contextmanager def get_session(self) -> Generator[Session, None, None]: session = self.session_factory() try: yield session session.commit() except Exception: session.rollback() raise finally: session.close() class Container(containers.DeclarativeContainer): db_context = providers.Factory( DatabaseContext, session_factory=db_session_factory ) user_repository = providers.Factory( UserRepository, db_context=db_context ) class UserRepository(UserRepositoryInterface): def __init__(self, db_context: DatabaseContext): self.db_context = db_context def create_user(self, user_data: UserCreate) -> User: with self.db_context.get_session() as db: user = User(**user_data.dict()) db.add(user) return user def get_by_email(self, email: str) -> Optional[User]: with self.db_context.get_session() as db: return db.query(User).filter(User.email == email).first()
Python
복사
이렇게 구성하면 FastAPI의 편의성, 클린 아키텍처의 원칙, 그리고 효율적인 데이터베이스 세션 관리를 모두 만족시킬 수 있다. 하지만 정말 자주 있지도 않을 데이터베이스 교체에 대비한다고 엄청난 보일러플레이트를 만들고 동작 원리를 동료들에게 설득하는 것도 정말 어려운 일이 아닐 수 없다. 프로젝트의 규모를 보고 결정하라.
parse me : 언젠가 이 글에 쓰이면 좋을 것 같은 재료을 보관해 두는 영역입니다.
1.
None
from : 과거의 어떤 원자적 생각이 이 생각을 만들었는지 연결하고 설명합니다.
1.
•
앞의 글에서 설명한 FastAPI의 Depends() 시스템은 분명 강력하고 테스트가 용이하며 편리하지만 클린 아키텍처의 모든 레이어에 적용하기에는 한계가 있다. 이건 컨트롤러(표현 계층)용이다.
2.
•
@를 이용한 데코레이터 문법을 사용하는 경우 코드가 간결해지지만 데코레이터가 데이터베이스 세션 생성 방식에 강하게 결합되어 테스트가 어려워진다는 한계가 있다.
supplementary : 어떤 새로운 생각이 이 문서에 작성된 생각을 뒷받침하는지 연결합니다.
opposite : 어떤 새로운 생각이 이 문서에 작성된 생각과 대조되는지 연결합니다.
1.
None
to : 이 문서에 작성된 생각이 어떤 생각으로 발전되거나 이어지는지를 작성하는 영역입니다.
2.
ref : 생각에 참고한 자료입니다.
1.
69p, FastAPI가 제공하는 Depends는 사용하기 편하다. 하지만 큰 단점이 있다. 우리는 클린 아키텍처로 소프트웨어를 만들고 있다. 따라서 의존성 역전은 필수 요소다. … UserService의 생성자가 호출될 때 Depends에 의해 의존성이 주입되는 UserRepository의 객체를 만들어야 하므로 Depends(UserRepository)가 될 수밖에 없다. 이는 애플리케이션 계층이 여전히 인프라 계층에 직접 의존한다는 뜻이다. … (IoC 컨테이너를 사용하면) 애플리케이션이 구동될 때 IoC 컨테이너에 미리 의존성을 제공하는 객체를 등록해두고 필요한 모듈에서 주입하도록 만들 수 있다. 이렇게 되면 주입할 때의 타입을 인터페이스로 선언하더라도 실제로 주입되는 객체는 구현체가 되도록 할 수 있게 된다. … (dependency-injector의) providers 모듈에는 Factory 외에도 여러 종류의 프로바이더를 제공한다. Factory는 개게를 매번 생성하고, Singleton은 처음 호출될 때 생성된 객체를 재활용한다.