[FastAPI] FastAPI 튜토리얼 (3)

Python/FastAPI

[FastAPI] FastAPI 튜토리얼 (3) - Dependency

ready-go 2023. 7. 20. 19:43

DI(Dependency Injection)

Dependency란 어떤 일을 진행하기 위해서 필요한 외부의 코드를 말한다. 시스템 차원에서 이러한 Dependency들을 필요한 곳에 제공하는 것을 DI(Dependency Injection)라고 한다. FastAPI도 DI를 간단히 사용할 수 있는 방법을 제공한다. 이번 글에서는 FastAPI에서 DI를 제공하는 방법에 대해 살펴본다.

참고로, Spring에서의 DI와 비교하자면 FastAPI에서의 DI는 조금 더 확장된 개념인 것 같다. Spring에서는 Bean으로 관리되는 컴포넌트들을 자동으로 주입해주는 기능이라면, 추가적으로 Filter에서 제공하는 기능을 일부 포함한다고 보면 될 것 같다.

Dependable

FastAPI에서 Dependency로 사용할 수 있는 것을 Dependable이라고 표현한다. 이러한 dependable은 함수처럼 호출할 수 있는 callable한 것이어야 한다. 가장 간단한 함수를 사용하는 예시를 보자.

Function

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

위 코드에서 common_parameters가 dependable에 해당한다. common_parameters는 이미 본 적이 있는 q라는 parameter를 받고 있다. Query parameter로 skip과 limit을 받을 때 사용했던 것과 똑같다. 그러나 common_parameters는 path operation 함수가 아니다. 대신, path operation 함수에 Depends(common_parameters)로 주입되고 있는 것을 확인할 수 있다.

위 코드를 Dependency가 없는 버전으로 바꾸면 아래와 같다.

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/users/")
async def read_users(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

동작은 똑같지만 코드의 재사용이 가능해졌다. 만약 똑같은 query parameter를 50개 함수가 사용한다면 따로 분리해 한 곳에서 관리하는 것이 유지, 보수에 훨씬 좋을 것이다.

이제, 조금 더 코드를 깔끔하게 해보자. Path operation 함수에 Annotated[dict, Depends(common_parameters)]라는 코드가 반복되고 있다. 별거 아닌 것 같지만, 아래 코드와 비교해보면 분리하는 것이 훨씬 깔끔하다는 것을 확인할 수 있다.

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}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
	return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
	return commons

Class

Dependable은 callable한 것이면 무엇이든 가능하다고 했다. class 역시 callable한 것이다.

class QueryParam:
	def __init__(self, q: str | None = None, skip: int, limit: int):
		self.q = q
		self.skip = skip
		self.limit = limit

예를 들어, 위와 같이 클래스가 정의되어있다면 아래와 같이 생성자를 호출할 수 있기 때문이다.

q = QueryParam(q="query", skip=10, limit=20)

이제 또 위에서 살펴본 코드와 같은 동작을 하는 코드를 만들 수 있다.

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class QueryParam:
	def __init__(self, q: str | None = None, skip: int, limit: int):
		self.q = q
		self.skip = skip
		self.limit = limit


@app.get("/items/")
async def read_items(query_params: Annotated[QueryParam, Depends(QueryParam)]):
	return query_params


@app.get("/users/")
async def read_users(query_params: Annotated[QueryParam, Depends(QueryParam)]):
	return query_params

Sub-dependency

지금까지 살펴본 Dependency를 여러개 연결해 사용할 수도 있다.

from typing import Annotated

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: str | None = None):
	return q


def query_or_cookie_extractor(
	q: Annotated[str, Depends(query_extractor)],
	last_query: Annotated[str | None, Cookie()] = None,
):
	if not q:
		return last_query
	return q


@app.get("/items/")
async def read_query(
	query_or_default: Annotated[str, Depends(query_or_cookie_extractor)]
):
	return {"q_or_cookie": query_or_default}

먼저 path operation 함수인 read_query를 보면 query_or_cookie_extractor에 의존하고있다. 따라서 /items/로 GET 요청이 오면 read_query 함수가 호출되기 전에 query_or_cookie_extractor 함수가 호출될 것이다.

query_or_cookie_extractor 함수는 또 다시 query_extractor 함수에 의존하고있다. 따라서 함수의 호출 순서는 아래와 같이 이뤄진다.

query_extractor
query_or_cookie_extractor
read_query

주입을 하지 않는 Dependency

지금까지는 Path operation 함수에 parameter를 주입해주는 예시만 살펴봤다. 그러나 반드시 무언가 주입해줄 필요는 없다. 단순히 callable한 dependable을 호출하는 것만으로도 충분하다.

이렇게 무언가 주입을 하지 않는 dependency는 path operation의 parameter가 아니라 decorator에 넣어줄 수 있다.

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


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.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
	return [{"item": "Foo"}, {"item": "Bar"}]

위 코드는 /items/로 들어온 GET 요청에 대해 HTTP Header의 값으로 권한을 검사하는 코드이다. 마치 Spring에서의 Filter의 역할과 빗슷하다. 권한 검사에 실패하면 exception을 일으켜 path operation 함수가 호출되지 않게 할 수 있다.

참고로, verify_token 함수는 반환값이 없고, verify_key는 반환값이 있다. 그러나 반환값의 유무와 관계없이 똑같이 decorator에 넣어준 것을 확인할 수 있다. 반환값이 있어도 사용하고싶지 않다면 위와 같이 사용할 수 있다.

Global Dependency

만약 FastAPI 서버로 들어오는 모든 요청에 대해 똑같은 권한 검사를 하고싶다면 모든 path operation 함수에 dependency를 명시해야 할 것이다. 그러나 이렇게 중복된 코드를 관리하는 것은 귀찮은 일이다.

FastAPI에서는 FastAPI 인스턴스를 생성할 때 global하게 dependency를 적용할 수 있는 방법을 제공한다.

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"}]

yield

FastAPI는 path operation이 반환되고 나서 추가적인 작업을 할 수 있는 방법을 제공한다. 이것은 dependable 함수에서 yield를 딱 한 번 사용하면 된다.

예를 들어, DB를 사용하는 API 요청에 대해 path operation 함수를 실행하기 전에 DB 세션을 열고, 응답을 보낸 후에 세션을 닫는 상황이 필요할 수 있다.

async def get_db():
	db = DBSession()
	try:
		yield db
	finally:
		db.close()
        

@app.patch("/users/{user_id}")
async def rename(user_id: int, user_name: str, db: Annotated[DBSession, Depends(get_db)]):
	return await user_service.rename(user_id, user_name, db)

Dependable을 위와 같이 정의하면 rename 함수가 호출되기 전 get_db가 호출되어 세션을 열고, rename이 반환되어 API 응답이 나갈 때까지 유지한다. rename이 응답을 보내고 나면 get_db의 yield 뒤의 코드가 실행된다. 따라서 자동으로 세션을 닫을 수 있다.

만약 rename에서 exception이 발생하는 경우 get_db에서 except로 추가적인 로직을 실행할 수도 있다. 그러나 주의할 점은 yield 뒤의 로직은 API 응답과는 관련이 없다는 것이다. 만약 yield 뒤에서 exception을 일으키거나 다른 응답을 보내고 싶어도 이미 응답을 보낸 후기 때문에 그럴 수 없다. 대신 DB 세션을 닫는 것처럼 응답과 관계없이 추가적인 작업을 하기 위해서만 사용할 수 있다.

3줄 요약

Dependency는 path operation 함수가 실행되기 전에 실행되는 추가적인 로직이다
반환값이 있으면 parameter에, 없으면 decorator에 넣어준다
yield를 사용하면 응답이 끝나고 추가적인 로직을 실행할 수도 있다