[FastAPI, Python] FastAPI의 의존성 구현과 주입 방법

FastAPI 스터디한 것 정리

Building Data Science Applications with FastAPI를 읽으며.

Chapter 5: 의존성 주입

의존성 주입

• 장점
  • 의도가 명확하다.
  • a clear separation of concern between the logic of the endpoint and the more generic logic
  • 명확한 관심의 분리가 가능하다
  • OpenAPI 스키마가 자동으로 제작되어서 어떤 파라미터가 필요한지 명확히 보여줄 수 있다.
    • can clearly show which parameters are expected

Creating and using a function dependency

• dependency는 function이나 callable class로 정의가 가능하다.
• dependency : 어떤 값이나 객체를 받아서 무엇인가 만들고, 그 무엇을 반환하는 로직을 wrapping 하는 방법.

async def pagination(skip: int = 0, limit: int = 10) -> Tuple[int, int]:

    return (skip, limit)

@app.get("/items")

async def list_items(p: Tuple[int, int] = Depends(pagination)):

    skip, limit = p

    return {"skip": skip, "limit": limit}

• pagination 함수: dependency를 정의함.
• list_items 함수: FastAPI에서 제공하는 Depends함수를 통해서 의존성을 주입함.
  • 함수를 인자로 받고, 실행시킴.
  • sub-dependencies들이 자동적으로 discover 되고 실행된다.
  • 한계점: parameter 선언시, 타입을 우리가 직접적으로 설정해주어야한다.(p: Tuple[int, int])
• TYPE HINT OF A DEPENDENCY RETURN VALUE

더욱 복잡한 활용

async def pagination(

    skip: int = Query(0, ge=0),

    limit: int = Query(10, ge=0),

) -> Tuple[int, int]:

    capped_limit = min(100, limit)

    return (skip, capped_limit)

• Query를 활용하여 최소값 설정 및 default 값 설정
• 최댓값 지정 로직 추가.

객체를 가져오거나 404 오류 발생시키기

get 엔드포인트는 원하는 객체를 가져오거나, 객체가 존재하지 않는 경우 404 오류를 발생시켜야 한다(update, delete에도 적용)

async def get_post_or_404(id: int) -> Post:

    try:

        return db.posts[id]

    except KeyError:

        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

이 예시의 핵심 포인트: dependency 내부에서도 error를 raise 할 수 있다.

• 이렇게 dependency를 이용해서 pre-conditions를 엔드포인트가 실행되기 전에 미리 확인, 검증할 수 있다.
  • Auth에서 유용

@app.get("/posts/{id}")

async def get(post: Post = Depends(get_post_or_404)):

    return post

@app.patch("/posts/{id}")

async def update(post_update: PostUpdate, post: Post = Depends(get_post_or_404)):

    updated_post = post.copy(update=post_update.dict())

    db.posts[post.id] = updated_post

    return updated_post

@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)

async def delete(post: Post = Depends(get_post_or_404)):

    db.posts.pop(post.id)

Parameterized dependency 만들기와 사용하기

• Dynamically limit을 설정하는 등의 좀 더 세밀한 구현이 필요한 경우
  • class를 활용할 것.
• Depends 함수의 요구 사항: being a callable
  • 즉, 클래스를 구현하고 내부에 __call__ 메서드를 구현할 것.
• 구현

class Pagination:
    def __init__(self, maximum_limit: int = 100):
        self.maximum_limit = maximum_limit

    async def __call__(
        self,
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=0),
    ) -> Tuple[int, int]:
        capped_limit = min(self.maximum_limit, limit)
        return (skip, capped_limit)

• 사용법

pagination = Pagination(maximum_limit=50)

@app.get("/items")

async def list_items(p: Tuple[int, int] = Depends(pagination)):

    skip, limit = p

    return {"skip": skip, "limit": limit}

여기서는 하드 코딩했지만, config file이나 environment variable로 동적인 변화 가능하다. 또한 클래스로 의존성을 정의하는 경우 로컬 값들이 메모리에 저장된다는 이점이 있다. 이는, 머신러닝 모델을 불러오는 등의 무거운 초기화 작업을 하는 경우에 유용하다.

클래스 메서드를 의존성으로 사용하기

__call__로 구현하는 것이 가장 직관적이지만, 클래스 내에 있는 다른 메서드도 callable 하기 때문에 의존성으로 사용하는 것이 가능하다. 이런 방법은 공통된 파라미터나 로직이 있는 경우 재사용성을 높이기에 좋다(pretrained ML model 등을 여러 곳에서 활용하는 경우).

Depends 함수에 넘겨 줄 때 class_name.class_method_name형태로 인자를 전달.

• Definition

class Pagination:

    def __init__(self, maximum_limit: int = 100):

        self.maximum_limit = maximum_limit

    async def skip_limit(

        self,

        skip: int = Query(0, ge=0),

        limit: int = Query(10, ge=0),

    ) -> Tuple[int, int]:

        capped_limit = min(self.maximum_limit, limit)

        return (skip, capped_limit)

    async def page_size(

        self,

        page: int = Query(1, ge=1),

        size: int = Query(10, ge=0),

    ) -> Tuple[int, int]:

        capped_size = min(self.maximum_limit, size)

        return (page, capped_size)

• Usage

pagination = Pagination(maximum_limit=50)

@app.get("/items")

async def list_items(p: Tuple[int, int] = Depends(pagination.skip_limit)):

    skip, limit = p

    return {"skip": skip, "limit": limit}

@app.get("/things")

async def list_things(p: Tuple[int, int] = Depends(pagination.page_size)):

    page, size = p

    return {"page": page, "size": size}

현재까지 요약

• 클래스로 의존성을 구현하는 경우(함수로 구현하는 것보다 더욱 심화된 방법), 아래의 케이스에서 유용하다
  • 동적으로 파라미터를 설정해주고 싶거나
  • 무겁고 오래 걸리는 init logic
  • 여러 의존성에서 재사용되는 로직

의존성을 path, router, global 수준에서 사용하기

의존성은 FastAPI에서 만들기를 권장하는 building blocks이다. 이는 특정 로직을 재사용하기 쉽게 하고, 코드 가독성을 높인다.

이때까지는 하나의 app 내에서 재사용했지만, 이를 path, router, global level에서 사용할 수 있다.

def secret_header(secret_header: str | None = Header(None)) -> None:

    if not secret_header or secret_header != "SECRET_VALUE":

        raise HTTPException(status.HTTP_403_FORBIDDEN)

Dependency를 path 데코레이터에 사용하기

@app.get("/protected-route", dependencies=[Depends(secret_header)])

async def protected_route():

    return {"hello": "world"}

Dependency를 라우터 전체에 사용하기

여러 endpoint에 한 번에 적용할 수 있다. 즉, 라우터 인스턴스를 처음 생성할 때, APIRouter class의 argument로 의존성 목록을 넣어준다.

• Case 1

router = APIRouter(dependencies=[Depends(secret_header)])

@router.get("/route1")

async def router_route1():

    return {"route": "route1"}

@router.get("/route2")

async def router_route2():

    return {"route": "route2"}

app = FastAPI()

app.include_router(router, prefix="/router")

• Case 2

router = APIRouter()

@router.get("/route1")

async def router_route1():

    return {"route": "route1"}

@router.get("/route2")

async def router_route2():

    return {"route": "route2"}

app = FastAPI()

app.include_router(router, prefix="/router", dependencies=[Depends(secret_header)])

Dependency를 application 전체에 사용하기

app = FastAPI(dependencies=[Depends(secret_header)])

@app.get("/route1")

async def route1():

    return {"route": "route1"}

@app.get("/route2")

async def route2():

    return {"route": "route2"}

어떤 level에서 의존성을 주입할 것인가?

요약

• Dependency Injection에 대해 공부
• Reusability without compromising code readability
  • retaining maximum readability
• retrieve and validate request param
• complex services performing ML tasks
• Class-based Dependency definition
  • Set dynamic params
  • keep local state
• 여러 계층에서 의존성 주입 가능
  • 전역
  • 라우터
    • include_router
    • 라우터 객체 생성시
  • Path
    • Decorator
    • Path operation function의 argument