WIL Dependency Injection
이번주 WIL은 FastAPI에서 의존성 주입을 어떻게 하는지 삽질의 기록이다.
Dependency Injection?
Dependency Injection(DI)이란? 소프트웨어 설계 패턴 중 하나로, 객체가 다른 객체의 의존성(즉, 객체가 필요로 하는 다른 객체나 리소스)을 직접 생성하지 않고, 외부(주로 프레임워크나 컨테이너)로부터 제공받는 방식이다.
왜 DI를 사용하는가
- 결합도 감소
객체가 의존하는 다른 객체에 대한 구체적인 지식 없이 인터페이스를 통해 상호 작용하므로, 시스템의 결합도가 낮아지고 각 컴포넌트를 독립적으로 개발하고 유지보수하기 쉬워진다. - 코드의 재사용성 향상
각 컴포넌트를 서로 다른 상황에서 재사용할 수 있으며, 특정 구현에 종속되지 않는다. - 테스트 용이성
실제 의존성을 테스트 용이성이 높은 모의 객체(mock objects)나 스텁(stubs)으로 대체할 수 있어, 단위 테스트가 훨씬 간편해진다.
FastAPI에서 어떻게 DI를 구현하는가?
FastAPI 공식문서에 자세히 나와있다.
1. api에 의존성 주입
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return commons
api endpoint paramter에 Depends를 사용함으로써, 의존성을 주입할 수 있다. 위 예제는 common_parameters 함수를 통해 공통 paramter를 정의하여 여러개의 api 에서 사용하고 있다.
2. 전역에 의존성 주입
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated
async def verify_token(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: Annotated[str, Header()]):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
@app.get("/items/")
async def read_items():
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
@app.get("/users/")
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
FastAPI를 생성할 때 parameter에 Depends를 사용함으로써 app 전체의 의존성을 추가할 수 있다.
3. yield를 통한 의존성 주입
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from .database import SessionLocal, Base, engine
# 데이터베이스 모델을 생성하는 부분
Base.metadata.create_all(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
app = FastAPI()
@app.get("/items/")
def read_items(db: Session = Depends(get_db)):
items = db.query(Item).all()
return items
db와 연결할 때에는 세션의 종료를 효율적으로 컨트롤 하기 위해 yield를 사용한다. get_db 함수는 SessionLocal()을 호출하여 새로운 데이터베이스 세션을 시작한다. read_items 함수에서 db 세션을 사용하여 데이터베이스에서 데이터를 조회한다. read_items 함수의 실행이 끝나고 반환값이 전달된 후, yield 다음에 위치한 finally 블록이 되고, 이 블록 내에서 db.close() 호출을 통해 데이터베이스 세션을 안전하게 종료할 수 있다.
간단하게 FastAPI 서버를 구축한다면 위와 같은 방법으로 충분히 의존성을 주입할 수 있다. 하지만 프로젝트가 커지면 효율적으로 관리하기 위해 Layerd Architecture를 사용하게 된다. 이때 위와 같은 방법으로는 의존성을 주입하는 것이 매우 번거로워진다. Python에서는 이러한 의존성 관리를 유연하게 하기 위해 다양한 DI 프레임워크가 존재하는데, 그 중 내가 사용한 것은 Dependency Injector 이다.
Dependency Injector
Dependency Injector 공식문서에 자세히 나와있다.
우선 Dependency Container를 만들어준다. 이 컨테이너는 Dependency Injector에서 의존성을 관리하는 기본 컨테이너이다.
AIResponseRepository의 생성자에서는 session_factory를 필요로 한다.
AIResponseService의 생성자에서는 AIResponseRepository를 필요로 한다.
각각 필요한 param에 따라 Container에 의존성을 주입해준다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# container.py
class Container(containers.DeclarativeContainer):
wiring_config = containers.WiringConfiguration(modules=["api.ai_response", "api.review"]) ## wiring 설정
db = providers.Singleton(Database, db_url=DATABASE_URL) ## Singleton 방식
ai_response_repo = providers.Factory( ## Factory 방식
AIResponseRepository,
session_factory=db.provided.session,
)
# ... 생략
ai_response_service = providers.Factory( ## Factory 방식
AIResponseService,
ai_response_repo=ai_response_repo,
)
# ... 생략
# repository.py
class AIResponseRepository:
def __init__(self, session_factory: Callable[..., AbstractContextManager[Session]]) -> None:
self.session_factory = session_factory
# ... 생략
# service.py
class AIResponseService:
def __init__(self, ai_response_repo: AIResponseRepository):
self.ai_response_repo = ai_response_repo
# ... 생략
위와 같은 방식으로 Container에 의존성에 대한 설정을 할 수 있다.
의존성 주입을 끝냈다면, wiring을 통해 컴포넌트 대상에 주입해줘야 한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# api
@router.get("/{ai_response_id}")
@inject
def get_ai_response(
ai_response_id: int,
ai_response_service: AIResponseService = Depends(Provide[Container.ai_response_service])
) -> AIResponseSchema:
ai_response = ai_response_service.find_by_id(ai_response_id)
return AIResponseSchema.from_orm(ai_response)
# main.py
# Container 인스턴스 생성
container = Container()
app = FastAPI()
Singleton Pattern
싱글톤 패턴은 특정 클래스의 인스턴스가 애플리케이션 전체에서 하나만 존재하도록 보장하는 디자인 패턴이다. 이는 전역 상태를 만들고, 모든 곳에서 쉽게 접근할 수 있도록 한다. 싱글톤은 주로 리소스 접근이 중복되지 않도록 제어해야 할 때 사용된다.
(예: 데이터베이스 연결, 로깅 메커니즘)