클린 아키텍처에서는 애플리케이션 계층(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 --> SupabaseUserRepositoryMermaid
복사
소스코드 입장에서 논리적으로는 컨테이너에 의존하는 모양새가 된다. 그 덕분에 컨테이너 내부에서는 마치 조물주가 피조물들에 영혼을 불어넣는 과정을 제어하듯, 어떤 구현체를 사용할지를 선택할 수 있어 소스코드의 강력한 중앙 관리자의 역할을 할 수 있게 된다.
마지막으로 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 : 어떤 새로운 생각이 이 문서에 작성된 생각을 뒷받침하는지 연결합니다.
1.
None
opposite : 어떤 새로운 생각이 이 문서에 작성된 생각과 대조되는지 연결합니다.
1.
None
to : 이 문서에 작성된 생각이 어떤 생각으로 발전되거나 이어지는지를 작성하는 영역입니다.
2.
ref : 생각에 참고한 자료입니다.