■ FastAPI 클래스에서 StreamingResponse 객체를 사용해 응답 스트리밍을 처리하는 방법을 보여준다.
▶ main.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
import json from pydantic import BaseModel from fastapi import APIRouter from fastapi import Path from fastapi.responses import StreamingResponse from fastapi import FastAPI class TodoItem(BaseModel): id : int item : str class Config: json_schema_extra = { "example" : { "id" : 1, "item" : "Example Schema!" } } class TodoItemUpdate(BaseModel): item : str class Config: json_schema_extra = { "example" : { "item" : "Read the next chapter of the book" } } todoItemList = [] apiRouter = APIRouter() @apiRouter.post("/todo") async def addTodoItem(todoItem : TodoItem) -> dict: todoItemList.append(todoItem) return {"message" : "Todo item added successfully"} @apiRouter.get("/todo") async def getTodoItemList() -> dict: return {"todoItemList" : todoItemList} async def enumerateTodoItemListInternal(): for todoItem in todoItemList: yield json.dumps(todoItem.dict()) + "\n" @apiRouter.get("/todos") async def enumerateTodoItemList(): return StreamingResponse(enumerateTodoItemListInternal(), media_type = "application/json") @apiRouter.get("/todo/{todoItemID}") async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict: for todoItem in todoItemList: if todoItem.id == todoItemID: return {"todoItem" : todoItem} return {"message" : "Todo item with supplied ID doesn't exist."} @apiRouter.put("/todo/{todoItemID}") async def updateTodo(todoItemUpdate: TodoItemUpdate, todoItemID : int = Path(..., title = "The ID of the todo item to be updated.")) -> dict: for todoItem in todoItemList: if todoItem.id == todoItemID: todoItem.item = todoItemUpdate.item return {"message" : "Todo item updated successfully."} return {"message" : "Todo item with supplied ID doesn't exist."} @apiRouter.delete("/todo/{todoItemID}") async def deleteTodo(todoItemID : int) -> dict: for index in range(len(todoItemList)): todoItem = todoItemList[index] if todoItem.id == todoItemID: todoItemList.pop(index) return {"message" : "Todo item deleted successfully."} return {"message" : "Todo item with supplied ID doesn't exist."} @apiRouter.delete("/todo") async def deleteAllTodoItems() -> dict: todoItemList.clear() return {"message" : "All Todo items deleted successfully."} fastAPI = FastAPI() @fastAPI.get("/") def root(): return {"message" : "Hello World"} @fastAPI.get("/home") def home(): return {"message" : "home"} fastAPI.include_router(apiRouter) |
▶ requirements.txt
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 |
annotated-types==0.6.0 anyio==4.3.0 certifi==2024.2.2 click==8.1.7 colorama==0.4.6 dnspython==2.6.1 email_validator==2.1.1 exceptiongroup==1.2.2 fastapi==0.111.0 fastapi-cli==0.0.3 h11==0.14.0 httpcore==1.0.5 httptools==0.6.1 httpx==0.27.0 idna==3.7 itsdangerous==2.2.0 Jinja2==3.1.4 markdown-it-py==3.0.0 MarkupSafe==2.1.5 mdurl==0.1.2 orjson==3.10.3 pydantic==2.7.1 pydantic-extra-types==2.7.0 pydantic-settings==2.2.1 pydantic_core==2.18.2 Pygments==2.18.0 python-dotenv==1.0.1 python-multipart==0.0.9 PyYAML==6.0.1 rich==13.7.1 shellingham==1.5.4 sniffio==1.3.1 starlette==0.37.2 typer==0.12.3 typing_extensions==4.11.0 ujson==5.10.0 uvicorn==0.29.0 uvloop==0.19.0 watchfiles==0.21.0 websockets==12.0 |
▶ 실행 명령
|
1 2 3 |
uvicorn main:fastAPI --reload |