[FastAPI Docs] Dependencies

JeongChaeJin·2022년 9월 1일

What is "Dependency Injection"

프로그램을 작동시키고, 사용하도록 하는 것을 선언하는 코드를 만들기 위한 방법이다.
공유된 logic
데이터베이스 coonection 공유
보안, 권한 등을 강화
code 반복을 줄이는 등등 의 이점들이 있댄다.

First Step

from typing import 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: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

common_parameters 함수를 보면 operation function과 동일한 형태이다.
Depends는 single parameter로 function을 보내줘야된다.
- 값을 받고, operation function이 동작하는 것과 동일하게 동작하게 된다.

FastAPI는 먼저 의존성 함수를 호출한다.
결과를 얻는다.
결과를 path operation function에 파라미터로 전달한다.
이는 path operation들 간에 공유되는 code를 쓸수있는 방법인 것이다.

Simple and Powerful

계층적인 의존성을 만들 때에도 유용하다.
/items/public/, /items/private/, /users/{user_id}/activate, /items/pro/ 이 4개 API 엔드포인트가 있다고 해보자.
Dependency Injection을 사용하면 이 서브 계층 순서대로 작동하는 API를 만들 수 있을 것이다.
- 엔드 포인트를 따라 공유되는 함수로 올라가는 것으로 이해된다.

Classes as Dependencies

from typing import 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: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

from typing import 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: 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

FastAPI에서 dependency는 Callable이다.
Class로 바꿔줘도 이전 function으로 만든 것과 동일하게 동작한다.
async def read_items(commons: CommonQueryParams = Depends()): 선언 시, 이렇게 shortcut도 가능하다.

Sub-dependencies

from typing import 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: str = Depends(query_extractor),
    last_query: Union[str, None] = Cookie(default=None),
):
    if not q:
        return last_query
    return q


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

dependency들은 또다른 dependency를 갖는 경우가 있다.

Dependencies in path operation decorators

dependency 에서의 값을 return 할 필요가 없는 경우도 있다.

Add dependencies to the path operation decorator

optional argument로 dependencies를 받는다.

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: 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: 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"}]

Global Dependencies

전체 어플리케이션에 의존성을 원할 수 있다.

from fastapi import Depends, FastAPI, Header, HTTPException


async def verify_token(x_token: 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: 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 with yield

FastAPI는 끝난 후 여분의 과정을 하는 의존성을 지원한다.

A database dependency with yield

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

Sub-dependencies with yield

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=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)