클린 아키텍처에서는 애플리케이션 계층(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 SQLAlchemyUserRepository(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()
Python
복사
하지만 FastAPI의 Depends()로는 이런 구조를 자연스럽게 구현하기 어렵다. UserService의 생성자에서 user_repo: UserRepositoryInterface = Depends()라고 작성해도, FastAPI는 인터페이스를 인스턴스화할 수 없다. 그렇다고 user_repo: UserRepositoryInterface = Depends(UserRepository) 라고 작성하는 순간, UserService가 UserRepository에 의존해 버리는 모순이 발생한다.
이 문제를 해결하려면 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 SQLAlchemyUserRepository
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(
SQLAlchemyUserRepository,
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의 의존성 주입 시스템에 등록한다. 참고로 @inject 데코레이터는 의존성이 주입되고 있음을 메모하는 표시일 뿐이다.
---
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
복사
---
title: After - IoC 컨테이너 사용 (의존성 역전 적용)
---
classDiagram
class LoginController {
+login(credentials)
}
class UserServiceInterface {
<<interface>>
+authenticate_user()
}
class UserService {
+authenticate_user()
-user_repo: UserRepositoryInterface
}
class UserRepositoryInterface {
<<interface>>
+get_by_email()
+create()
}
class SQLAlchemyUserRepository {
+get_by_email()
+create()
-db: Session
}
class Container {
+user_service: Factory
+user_repository: Factory
}
LoginController --> UserServiceInterface : Depends(Provide[Container.user_service])
UserService ..|> UserServiceInterface : implements
UserService --> UserRepositoryInterface : 생성자에서 인터페이스 의존
SQLAlchemyUserRepository ..|> UserRepositoryInterface : implements
Container ..> UserService : creates
Container ..> SQLAlchemyUserRepository : creates
note for LoginController "컨트롤러는 인터페이스에만 의존"
note for Container "컨테이너가 구체 구현체 결정"Mermaid
복사
IoC 컨테이너를 사용한다고 완벽한 의존성 역전이 일어나는 것은 아니다. 어떤 구현체를 사용할지 컨테이너에서 결정할 수 있다는 점을 고려할 때, 중앙화로서의 역할이 더 강력하다.
마지막으로 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 컨테이너를 사용하는 경우, 테스트 의존성은 다음과 같이 교체할 수 있다.
# tests/conftest.py
@pytest.fixture
def container():
container = Container()
# 테스트용 설정
container.config.database_url.override("sqlite:///test.db")
# 필요시 특정 의존성만 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 SQLAlchemyUserRepository(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
복사
하지만 데코레이터가 데이터베이스 세션 생성 방식에 강하게 결합되어 테스트 시 DB를 교체하는 것이 어려워진다. 더 유연한 방법은 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)
)
db_session_factory = providers.Singleton(
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(
SQLAlchemyUserRepository,
session_decorator=db_session_decorator
)
Python
복사
그리고 리포지토리에서는 주입받은 데코레이터를 사용한다.
class SQLAlchemyUserRepository(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
복사
이 방법은 IoC와 잘 통합되지만 코드 이해가 어렵고 복잡하다. 더 pythonic한 방법으로는 컨텍스트 매니저를 사용하는 방법이다.
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(
SQLAlchemyUserRepository,
db_context=db_context
)
class SQLAlchemyUserRepository(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 : 어떤 새로운 생각이 이 문서에 작성된 생각을 뒷받침하는지 연결합니다.
1.
None
opposite : 어떤 새로운 생각이 이 문서에 작성된 생각과 대조되는지 연결합니다.
1.
None
to : 이 문서에 작성된 생각이 어떤 생각으로 발전되거나 이어지는지를 작성하는 영역입니다.
ref : 생각에 참고한 자료입니다.