[Product Serving Part.5] FastAPI 기초-3

1. Event Handler

이벤트가 발생했을 때, 그 처리를 담당하는 함수

FastAPI에선 Application이 실행할 때, 종료될 때 특정 함수를 실행할 수 있음

 

from fastapi import FastAPI
import uvicorn

# Fastapi 객체 생성
app = FastAPI()

items = {}

@app.on_event("startup")
def startup_event():
    print("startup event")
    items["foo"] = {"name": "Fighters"}

@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)

 

2. API Router

API Router는 여러 API를 연결해서 활용한다고 생각하면 된다.

기존에 사용하던 @app.get, @app.post를 사용하지 않고, router 파일을 따로 설정하고 app에 import해서 사용

 

• include_router를 이용해 app에 연결
• 보통의 경우 각각의 파일에 별도 저장해서 사용 (user.py)

from fastapi import FastAPI, APIRouter
import uvicorn

# Router 객체 생성
user_router = APIRouter(prefix="/users")

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

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

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

# Fastapi 객체 생성
app = FastAPI()


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

 

 

 

예제 프로젝트 구조

 

 

 

3. Error Handling

• Error Handling은 웹 서버를 안정적으로 운영하기 위해 반드시 필요한 주제
• 서버에서 Error가 발생한 경우, 어떤 Error가 발생했는지 알아야 하고 요청한 클라이언트에 해당 정보를 전달해 대응할 수 있어야 함
• 서버 개발자는 모니터링 도구를 사용해 Error Log를 수집해야 함
• 발생하고 있는 오류를 빠르게 수정할 수 있도록 예외 처리를 잘 만들 필요가 있음

 

example) item_id가 4 이상의 숫자가 들어올 경우 Key Error 발생 (Internal Server Error)

from fastapi import FastAPI
import uvicorn


# Fastapi 객체 생성
app = FastAPI()

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

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


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

 

item_id가 4일 경우 Interneal Server Error가 return됨

이렇게 되면 클라이언트는 어떤 에러가 난 것인지 정보를 얻을 수 없음

에러 메세지와 에러의 이유 등을 클라이언트에 전달하도록 작성해야 함

 

FastAPI의 HTTPException은 Error Response를 쉽게 보낼 수 있게 해주는 Class

from fastapi import FastAPI, HTTPException
import uvicorn


# Fastapi 객체 생성
app = FastAPI()

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

@app.get("/v1/items/{item_id}")
async def get_item(item_id: int):
    try:
        item = items[item_id]
    except KeyError as e:
        raise HTTPException(
            status_code=404, detail=f"Item not found [id: {item_id}]"
        ) from e
    return item

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

 

 

4. Background Task

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

 

import contextlib
import json
import threading
import time
from datetime import datetime
from time import sleep
from typing import List
from uuid import UUID, uuid4

import requests
import uvicorn
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field

class Server(uvicorn.Server):
    def install_signal_handlers(self):
        pass

    @contextlib.contextmanager
    def run_in_thread(self):
        thread = threading.Thread(target=self.run)
        thread.start()
        try:
            while not self.started:
                time.sleep(1e-3)
            yield
        finally:
            self.should_exit = True
            thread.join()


def run_tasks_in_fastapi(app: FastAPI, tasks: List):
    """
    FastAPI Client를 실행하고, task를 요청합니다
    Returns:
        List: responses
    """
    config = uvicorn.Config(app, host="127.0.0.1", port=5000, log_level="error")
    server = Server(config=config)
    with server.run_in_thread():
        responses = []
        for task in tasks:
            response = requests.post("http://127.0.0.1:5000/task", data=json.dumps(task))
            if not response.ok:
                continue
            responses.append(response.json())
    return responses

# Fastapi 객체 생성
app = FastAPI()

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

task_repo = {}


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


@app.post("/task", status_code=202) # 비동기 작업이 등록되었을 때 202 상태 코드를 반환
async def create_task_in_background(task_input: TaskInput, background_task: BackgroundTasks):
    # 비동기 작업을 등록
    background_task.add_task(cpu_bound_task, id_=task_input.id_, wait_time=task_input.wait_time)
    return task_input.id_

@app.get("/task/{task_id}")
def get_task_result(task_id: UUID):
    try:
        return task_repo[task_id]
    except KeyError:
        return {"detail": "Task not found"}


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

 

 

프로젝트 구조 - Cookiecutter

cookitcutter-fastapi https://github.com/arthurhenrique/cookiecutter-fastapi