Dependencies

https://fastapi.tiangolo.com/tutorial/dependencies/

Dependency Injection은 두 가지로 분해해서 의미를 해석할 수 있는데 Dependency는 개발자의 code를 위해 사용하거나, 동작하기를 원하는 것을 선언하는 방식이다. Injection은 개발자의 code에게 필요한 dependency들을 제공해주는 어떤 것들을 관리하는 것을 의미한다. 즉 이 둘을 합치면 개발을 하는데 있어 필요한 dependency를 어떻게 system이 injection해주고, 관리해줄 것인가이다.

Dependency Injection은 다음과 같은 장점이 있다.
1. 공통괸 logic을 재사용할 수 있다.
2. database connection을 공유할 수 있다.
3. 보안적인 모듈들을 적용하기 쉽게 할 수 있다.

그럼 어떻게 FastAPI에서 dependency injection을 제공하고 있는 지 알아보도록 하자.

FastAPI의 ''Dependency' 또는 'Dependable' 만들기

Depends()라는 함수로 쓰이고 Query(), Body(), Form()와 사용법이 같다. Depends()안에는 하나의 함수가 들어갈 수 있는데, 이 함수는 handler에서 받는 parameter들과 동일한 parameter들을 받을 수 있다.

다음의 예시를 보도록 하자.

from typing import Annotated, Union
from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(
    q: Union[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

fastapi모듈의 Depends를 handler에 사용하되 handler에 하나의 함수를 전달해준다. 이 함수 안에서 개발자가 원하는 처리가 이루어지고, 그 결과를 Dependency로 handler에 전달하는 것이다. 위에서는 commons가 dependency가 된다.

그렇다면 Depends()에 들어가는 함수는 무엇인가?? 자세히 보면 handler(path operation function)과 크게 다르지 않아보인다. HTTP request를 통한 입력값들을 받고, 이를 선제적으로 처리하여 json으로 반환하는 것이 전부이다. 즉, common_parameters는 handler와 크게 다르지 않으며 path operation decorator만 없는 형식으로 볼 수 있다.

참고로, dependency의 모든 요청 선언, 검증, 요구사항들은 OpenAPI schema에 함께 결합된다. 따라서, OpenAPI docs로 보면 Depends를 사용해도 그닥 크게 달라보이는 것이 없다.

classes as Dependencies

위의 code에서는 commons의 type을 dict로 잡아놓았기 때문에 editor에서 자동완성이 불가능하다.

common_parameters는 dependency인데 위의 예제에서는 하나의 함수로 적혀있다. 사실 dependency의 조건은 함수라기 보다는 callable이다. 즉 __call__을 구현한 모든 것들은 dependency로 쓰일 수 있는 것이다. 따라서 class역시도 __call__이 있기 때문에 하나의 dependencies로 볼 수 있다.

class의 __call__은 class의 인스턴스를 생성할 때 사용된다. 가령 Something이라는 class가 있다면 Something()으로 call을 하게되는 것이다.

위의 dependency로 함수를 썼을 때를 보면 handler(path operation function)처럼 FastAPI에서 parameter를 자동으로 할당해준다. class역시도 마찬가지로 class의 인스턴스를 생성하는 __init__에 해당 parameter들이 들어가는 것이다.

from typing import Annotated, Union
from fastapi import Depends, FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

class CommonQueryParams:
    def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

__init__에 common_parameters function의 parameter처럼 FastAPI request parameter들이 자동으로 할당되어 인스턴스를 만드는 것이다. 그리고 해당 인스턴스가 바로 commons에 들어가는 것이다.

이렇게 class를 쓰든 function을 쓰든 OpenAPI에서는 동일하게 나온다.

단, 이렇게 class로 만들어쓰게 된다면 editor에서 자동완성 기능이 제공되고, 타입 추론이 훨씬 용이하기 때문에 유지 보수에 더 좋은 측면이 있다.

또한, 다음과 같이 Depends를 간단하게 바꾸어 쓸 수 있다. 단, 이 경우는 dependency가 class인 경우만 가능하다.

commons: Annotated[CommonQueryParams, Depends()]

그렇다면 여러 Depends들을 연결하여 하나의 pipeline을 만드는 것도 가능하지 않을까? sub-dependency를 구성하는 것도 가능하다.

Sub-dependencies

Depends를 이용하여 dependency안에 또 다른 sub-dependency를 두도록 하여 deep한 pipe를 만들 수 있다.

from typing import Annotated, Union
from fastapi import Cookie, Depends, FastAPI

app = FastAPI()

def query_extractor(q: Union[str, None] = None):
    return q

def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[Union[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}

다음의 예시를 보도록 하자.

read_query에 query_or_default라는 parameter를 두고 Depends로 의존성을 설정했다. query_or_cookie_extractor는 의존성 함수로 이 내부역시 Depends로 query_extractor를 의존성으로 두고 있다. 이는 sub-dependency를 설정한 것이다.

재밌는 것은 query_or_cookie_extactor 함수에서는 q parameter에 관해서는 query_extractor에 의존하고 있지만, last_query는 Cookie로 그대로 받아내고 있다. 이는 특정 parameter는 Depends로 관리하지만 특정 parameter는 그대로 받아내는 것이 가능하다는 것이다. FastAPI에서 자동으로 처리해주고 있는 것을 알 수 있다.

만약, 하나의 handler에 여러 dependency들이 붙어있고, 이 내부적으로 의존성들이 동일한 sub-dependency를 사용하고 있다면, fastapi는 request당 오직 한 번만 sub-dependency를 호출한다.

def query_extractor(q: Union[str, None] = None):
    print("call count")
    return q

def default_query_extractor(
    q: Annotated[str, Depends(query_extractor)]
):
    return q

def query_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[Union[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(default_query_extractor)],
    query_or_default2: Annotated[str, Depends(query_cookie_extractor)]
):
    return {"q_or_cookie": query_or_default, "q_or_cookie2": query_or_default2}

다음의 코드를 보면, read_query handler는 default_query_extractor와 query_cookie_extractor dependency를 가지고 있다. 각각의 dependency는 동일한 query_extractor sub-dependency를 가지고 있는데, sub-dependency인 query_extractor가 한 번 씩 호출되어 두 번 호출될 것 같아보이지만, 실제로는 한번만 호출되고 cache에 저장되어 두번째에는 호출되지 않고 결과만 반환한다. 따라서 위의 code에 요청을 보내면 call count log가 한번만 찍히게 된다. 단 캐싱되는 것은 해당 request일 때만 캐싱되고, 해당 request가 끝나면 캐싱은 완료된다.

만약, 이렇게 sub-dependency가 캐싱되지않고 매번 호출되도록 하고 싶다면 Depends에 use_cache=False로 두어야 캐싱되지 않고 매번 호출된다.

Dependencies in path operation decorators

dependency를 handler의 path operation function으로 쓰고 싶지만 딱히 return value로 낼 값이 없을 때가 있다. 가령, 해당 handler에 인증, 인가된 user만 접근이 가능한 경우가 대표적이다.

이런 경우 path operation에 Depends를 사용하는 것이 아니라 handler의 decorator에 dependencies를 활용하여 사용하면 된다. 참고로 dependencies는 list형식으로 여러 개의 dependency function을 차례대로 사용할 수 있다.

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

read_items handler의 decorator로 dependencies가 있는 것을 볼 수 있다. 여기에 list형식으로 Depends() 의존성들을 넣어주면 된다. 이들은 순서대로 실행되며, 각 dependency의 return값은 의미가 딱히 없다. 단, raise로 발생시킨 exception은 handler로 전달이 된다.

Global Dependencies

특정 dependencies는 관련된 모든 handler에 적용되어야 할 때가 있다. FastAPI 객체를 통해 관련된 모든 handler에 dependencies를 적용시킬 수 있는데, 다음을 보도록 하자.

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에 dependencies로 Depends(verify_token), Depends(verify_key)를 주입한 것을 알 수 있다. 이들을 통해서 FastAPI의 app`에 라우팅되는 모든 handler는 해당 dependency를 거치게 되는 것이다.

이는 또한, 추후에 APIRouter를 통해서 sub router를 구현할 때도 마찬가지이다. APIRouter역시도 dependencies를 설정할 수 있어 관련된 API에 들에 대해서 dependency를 injection시킬 수 있다.

Dependencies with yield

dependency는 handler가 실행되기 전 뿐만아니라, 끝난 후에도 실행될 수 있도록 하는 로직이 있다. 이는 yield를 통해서 이용할 수 있다.

가령, database session을 만들고 사용한 이후에 이를 close하는 code를 만들 수 있는데, dependency에서 yield를 기점으로 열고, 닫기를 표시하면 된다.

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

yield가 실행된 db connection은 handler의 path operation에 따라 dependency가 주입된다. 즉, parameter에 들어간다. handler가 response를 반환한 후에는 yield의 다음 줄이 실행되어 db.close를 실행한다.

또한, yield와 함께 try block을 사용한다면 handler에서 발생한 어떠한 exception이라도 받아낼 수 있다.

sub dependency에 yield를 사용할 수도 있다. 하나의 chain처럼 sub dependency구조를 만들 수 있는데, FastAPI에서는 exit code가 각 dependency마다 yield 동작하도록 보장하기 때문이다.

다음의 예제는 dependency_c가 dependency_b를 의존성으로 갖고, dependency_b가 dependency_a를 의존성으로 갖는 모양새이다.

from typing import Annotated
from fastapi import Depends

async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()

async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)

async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

dependency_c의 경우 exit code를 실행하기 위해서는 dependency_b로부터 값을 요구한다. 다음으로 dependency_b는 dependency_a를 요구하고 있다.

dependency_c가 종료되면 dependency_b의 finllay부분이 실행되고 dependency_b가 종료되면 dependency_c의 finally부분이 실행된다.

이는 사실 python의 Context Manager덕분인 것이다.

yield를 사용하여 handler의 exception도 받아내어 응답을 처리할 수 있다. 단, 이 경우는 FastAPI, 0.106.0이상만 가능하다.

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

data = {
    "plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},
    "portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}

class OwnerError(Exception):
    pass

def get_username():
    try:
        yield "Rick"
    except OwnerError as e:
        raise HTTPException(status_code=400, detail=f"Owner error: {e}")

@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id not in data:
        raise HTTPException(status_code=404, detail="Item not found")
    item = data[item_id]
    if item["owner"] != username:
        raise OwnerError(username)
    return item

참고로 custom exception handler를 만들어서 exception을 raise하고 처리할 수도 있다. 즉, dependency에서도 custom exception을 받아내고 custom exception handler를 마지막에 동작하도록 할 수 있다는 것이다.

Context Managers

마지막으로 yield가 어떻게 동작하는 지 알아보도록 하자.

Context Managers는 python object로 with statemnet와 함께 사용할 수 있는 object를 말한다.

with open("./somefile.txt") as f:
    contents = f.read()
    print(contents)

open()함수를 통해서 만들어진 객체를 Context Manager라고 부르는 것이다. 여기에서는 file 객체이기 때문에 with문이 끝난 뒤에는 file에 대한 session을 닫아버린다. 심지어 exception이 발생해도 file에 대한 session을 닫아버린다.

사실 Context Manager의 내부 동작은 매우 간다한데, python object가 __enter__과 __exit__이라는 두 method를 구현하면 with으로 시작할 때 __enter__를 실행하고 끝날 때 __exit__을 실행하는 것이 전부이다.

dependency를 yield와 함께 사용하면 FastAPI는 내부적으로 context manager를 만들어주고 다른 관련된 tool들과 연결시켜준다.