[FastAPI] 파일 분할

 

🐰 FastAPI 공식문서를 보면서 개인적으로 정리한 글 입니다.

 

FastAPI로 코드를 작성할 때 기능이 많고 규모가 크면 코드 관리가 힘들다. 이때는 파일을 분할하여 관리하는 것이 바람직하다.

 

다음과 같이 파일을 분리한다. 

 

 

routers - API 엔드포인트

from fastapi import APIRouter

router = APIRouter()


@router.get("/users/", tags=['users'])
async def read_items():
    return [{'username': 'Abdule'}, {'username': 'Bubi'}, {'username': 'Collson'}]


@router.get("/users/me", tags=['users'])
async def read_users_me():
    return {'username': 'Pupba'}


@router.get("/users/{username}", tags=['users'])
async def read_users(username: str):
    return {'username': username}

 

APIRouter 클래스를 사용하여 API 엔드포인트를 선언하고 router 인스턴스를 FastAPI app에 등록할 수 있다.

 

 

Dependencies - 종속성 정의

from typing import Annotated
from fastapi import Header, HTTPException


async def get_token_header(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(
            status_code=400, detail="No Jessica token provided")

 

종속성을 정의하여 불러와 사용할 수 있다.

import json
from fastapi import APIRouter, Depends, HTTPException
from ..dependencies import get_token_header
from typing import Dict

router = APIRouter(
    prefix='/items',
    tags=['items'],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}}
)

path = "./routers/fake_db.json"
with open(path) as f:
    json_object = dict(json.load(f))['items']


def findItem(target: str) -> dict:
    result: dict = {}
    for i in json_object:
        if list(i.values())[0] == target:
            result = i
            break
    return result


@router.get("/items/", response_model=Dict[str, str])
async def read_items():
    return json_object


@router.get("/{item_name}/", response_model=Dict[str, str])
async def read_item(item_name: str):
    result: dict = findItem(item_name)
    if result == {}:
        raise HTTPException(status_code=404, detail="Item not found")
    return result


@router.put(
    "{item_name}",
    tags=['custom'],
    responses={403: {"description": "Operation forbidden"}}
)
async def update_item(item_name: str):
    result: dict = findItem(item_name)
    if {} == result:
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus")
    return {item_name: "Banana"}

 

APIRouter에 매개변수를 설정하여 엔드포인트에 공통적으로 적용할 수 있다.

• prefix : APIRouter에 속한 엔드포인트들의 경로를 설정, ex. '/items' -> '/items/{item_name}
• tags : APIRouter에 속한 엔드포인트들의 tags를 설정, 엔드포인트 그룹화에 사용
• dependencies : APIRouter에 속한 엔드포인트들의 의존성을 설정
• responses : APIRouter에 속한 엔드포인트들의 응답에 대한 설정

 

 

Main - 메인 APP

앞에 만든 모듈들을 실행할 메인 APP을 정의하여 서버를 실행한다.

그 전에 internal 폴더를 만들어 다른 프로젝트와 공유하는 엔드포인트 정의 파일을 선언하고 메인에서 수정하여 사용할 수 있다.

이렇게 하면 기존 파일을 수정하지 않고 사용할 수 있다.

from fastapi import APIRouter

router = APIRouter()

@router.post("/")
async def update_admin():
    return {"message": "Admin getting schwifty"}

 

이렇게 정의한 파일을 main에서 prefix, tags 등을 수정하여 사용한다.

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

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

app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

 

app의 include_router를 통해 엔드포인트를 정의한 router 인스턴스를 추가할 수 있다.

또 admin 부분처럼 매개변수를 지정하여 internal 엔드포인트에 대한 설정을 할 수 있다.