Fast API (3)

<목차>
1. Event Handler
2. API Router
3. Error Handling
4. Background Task
5. Cookiecutter

 

Event Handler

- 이벤트가 발생했을 때, 그 처리를 담당하는 함수
- FastAPI에선 Application이 실행할 때, 종료될 때 특정 함수를 실행할 수 있음

Event Handler 사용 예시
- startup 할 때 머신 러닝 모델 load
- shutdown 할 때 로그 저장

- @app.on_event(“startup”)
- @app.on_event(“shutdown”)

from fastapi import FastAPI
import uvicorn

app = FastAPI()

items = {}

# ⭐ 데코레이터로 startup할 때 실행된다고 알려줌
@app.on_event("startup")
def startup_event():
    print("Start Up Event")
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}
# ⭐

@app.on_event("shutdown")
def shutdown_event():
    print("Shutdown Event!")
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")


@app.get("/items/{item_id}")
def read_items(item_id: str):
    return items[item_id]


if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8000)

---

API Router

- 기존에 사용하던 @app.get, @app.post을 사용하지 않고, router 파일을 따로 설정하고 app에 import해서 사용
- APIRouter는 Mini FastAPI로 여러 API를 연결해서 활용

from fastapi import FastAPI, APIRouter
import uvicorn

user_router = APIRouter(prefix="/users")
order_router = APIRouter(prefix="/orders")

# ⭐ user router와 order router 두 개를 생성한다
@user_router.get("/", tags=["users"])
def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


@user_router.get("/me", tags=["users"])
def read_user_me():
    return {"username": "fakecurrentuser"}


@user_router.get("/{username}", tags=["users"])
def read_user(username: str):
    return {"username": username}


@order_router.get("/", tags=["orders"])
def read_orders():
    return [{"order": "Taco"}, {"order": "Burritto"}]


@order_router.get("/me", tags=["orders"])
def read_order_me():
    return {"my_order": "taco"}


@order_router.get("/{order_id}", tags=["orders"])
def read_order_id(order_id: str):
    return {"order_id": order_id}


app = FastAPI()

if __name__ == '__main__':
    app.include_router(user_router) # ⭐ app에 연결한다
    app.include_router(order_router)
    uvicorn.run(app, host="0.0.0.0", port=8000)

실제로 활용한다면 위 코드처럼 하나의 파일에 저장하지 않고 user.py, order.py 두 개 파일에 각각 저장한다.

 

예제 프로젝트 구조

---

Error Handling

- 서버에서 Error가 발생한 경우, 어떤 Error가 발생했는지 알아야 하고 요청한 클라이언트에 해당 정보를 전달해 대응할 수 있어야 함
- 서버 개발자는 모니터링 도구를 사용해 Error Log를 수집해야 함
- 발생하고 있는 오류를 빠르게 수정할 수 있도록 예외 처리를 잘 만들 필요가 있음


- FastAPI의 HTTPException은 Error Response를 더 쉽게 보낼 수 있도록 하는 Class
- HTTPException을 이용해서 클라이언트에게 더 자세한 에러 메시지를 보내는 코드 작성

''' #https://github.com/jjeongah/Boostcamp-AI-Tech-Product-Serving/blob/main/part3/01-fastapi/examples/13_exception_handling.py
- item_id가 1~3까진 정상
- 4 이상의 숫자가 들어올 경우 Key Error가 발생
- 5인 경우 Internal Server Error
'''
from fastapi import FastAPI, HTTPException
import uvicorn

app = FastAPI()

items = {
    1: "Boostcamp",
    2: "AI",
    3: "Tech"
}

@app.get("/v1/{item_id}")
async def find_by_id(item_id: int):
    return items[item_id]

@app.get("/v2/{item_id}")
async def find_by_id(item_id: int):
    try:
        item = items[item_id]
    except KeyError:
        raise HTTPException(status_code=404, detail=f"아이템을 찾을 수 없습니다 [id: {item_id}]")
    return item

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8000)

---

Background Task

- FastAPI는 Starlett이라는 비동기 프레임워크를 래핑해서 사용
- FastAPI의 기능 중 Background Tasks 기능은 오래 걸리는 작업들을 background에서 실행함
- Online Serving에서 CPU 사용이 많은 작업들을 Background Task로 사용하면, 클라이언트는 작업 완료를 기다리지 않고 즉시 Response를 받아볼 수 있음

- Background Tasks를 사용하지 않은 작업들은 작업 시간 만큼 응답을 기다림

app_1 = FastAPI()

def cpu_bound_task(wait_time: int):
    sleep(wait_time)
    return f"task done after {wait_time}"

class TaskInput(BaseModel):
    wait_time: int = Field(default=1, le=10, ge=1)

@app_1.post("/task")
def create_task(task_input: TaskInput):
    return cpu_bound_task(task_input.wait_time)

tasks = [{"wait_time": i} for i in range(1, 10)]

start_time = datetime.now()
run_tasks_in_fastapi(app_1, tasks)
end_time = datetime.now()
print(f"Simple Tasks: Took {(end_time - start_time).seconds}")

- Background Tasks를 사용한 작업들은 기다리지 않고 바로 응답을 주기 때문에 0초 소요
- 실제 작업은 Background에서 실행됨

# 2. background tasks
app_2 = FastAPI()

@app_2.post("/task",
            status_code=202)  # 비동기 작업이 등록됐을 때, HTTP Response 202 (Accepted)를 보통 리턴합니다. https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/202
async def create_task_in_background(task_input: TaskInput, background_tasks: BackgroundTasks):
    background_tasks.add_task(cpu_bound_task, task_input.wait_time)
    return "ok"

start_time = datetime.now()
run_tasks_in_fastapi(app_2, tasks)
end_time = datetime.now()
print(f"Background Tasks: Took {(end_time - start_time).seconds}")

 

- 작업 결과물을 조회할 때는 Task를 어딘가에 저장해두고, GET 요청을 통해 Task가 완료됐는지 확인

# 3. background tasks with in-memory task repo
from uuid import UUID, uuid4

app_3 = FastAPI()

class TaskInput2(BaseModel):
    id_: UUID = Field(default_factory=uuid4)
    wait_time: int

task_repo = {}

def cpu_bound_task_2(id_: UUID, wait_time: int):
    sleep(wait_time)
    result = f"task done after {wait_time}"
    task_repo[id_] = result

@app_3.post("/task", status_code=202)
async def create_task_in_background_2(task_input: TaskInput2, background_tasks: BackgroundTasks):
    background_tasks.add_task(cpu_bound_task_2, id_=task_input.id_, wait_time=task_input.wait_time)
    return task_input.id_

@app_3.get("/task/{task_id}")
def get_task_result(task_id: UUID):
    try:
        return task_repo[task_id]
    except KeyError:
        return None

---

Cookiecutter

GitHub - cookiecutter/cookiecutter: A cross-platform command-line utility that creates projects from cookiecutters (project templates), e.g. Python package projects, C projects.

GitHub - arthurhenrique/cookiecutter-fastapi: Cookiecutter template for FastAPI projects using: Machine Learning, Poetry, Github Actions and Pytests

- CLI 형태로 프로젝트 생성 과정을 도와줌
- 회사에서 공통의 프로젝트 구조가 필요하면 쿠키 커터로 설정
- 개인용 쿠키 커터 템플릿을 만드는 것도 좋은 방법