[FastAPI] 1. GET, POST, PUT, DELETE ๋ฑ ๊ธฐ๋ณธ API ๋ง๋ค๊ณ ๋ฌธ์ ์๋ํ ํด๋ณด๊ธฐ
์ง๋ ๊ธ์ ์ด์ด์ FastAPI๋ฅผ ์ด์ฉํด์ ๊ธฐ๋ณธ์ ์ธ REST API ๊ฐ๋ฐ ์ด์ผ๊ธฐ ๋ ๋ฒ์งธ๋ฅผ ์ด์ด๊ฐ๋ณด๋๋ก ํ๊ฒ ์ต๋๋ค.
์ฐ๋ฆฌ๊ฐ REST API๋ฅผ ๊ฐ๋ฐํ๋ฉด ์ฃผ์ ๋ฉ์๋ ์ข ๋ฅ 5๊ฐ์ง๊ฐ ์๋๋ฐ์. ๋ฐ๋ก GET, POST, PUT, DELETE, PATCH ์ด๋ ๊ฒ 5๊ฐ์ง ์ ๋๋ค. ๋ฌผ๋ก ์ด ์ธ์๋ OPTIONS๊ฐ ์์ง๋ง ์ด ๊ธ์์๋ ๋ค๋ฃจ์ง ์๊ฒ ์ต๋๋ค.
๊ธฐ๋ณธ API ๋ง๋ค๊ธฐ (GET)
REST API์ ๊ธฐ๋ณธ API๋ ๋จผ์ GET ๋ฉ์๋๋ถํฐ ์์๋ฉ๋๋ค. GET ๋ฉ์๋๋ ์๋ฒ๋ก๋ถํฐ ๋ฆฌ์์ค๋ฅผ ๊ฐ์ ธ์ฌ ๋ ์ฐ๋ READ API ์ ๋๋ค. FastAPI์์๋ ์๋์ ๋ ๊ฐ์ง ๋ฐฉ๋ฒ์ผ๋ก ์์ฑํ ์ ์์ต๋๋ค.
from fastapi import FastAPI
app = FastAPI()
@app.route('/health')
async def health_check():
return "OK"
import uuid
from fastapi import FastAPI
from starlette.responses import JSONResponse
app = FastAPI()
@app.get("/{name}")
async def generate_id_for_name(name: str):
return JSONResponse({
'id': str(uuid.uuid4()),
'name': name
})
์ฒซ ๋ฒ์จฐ ๋ฐฉ๋ฒ์ route ๋ฉ์๋๋ฅผ ์ด์ฉํ๋ ๋ฐฉ๋ฒ์ ๋๋ค. route์ ๋ฉ์๋๋ฅผ ์ด์ฉํ ๊ฒฝ์ฐ ์ค์ง String์ ํด๋นํ๋ ๋ฐ์ดํฐ ํ์ ๋ง์ ๋ฐํ ์ํฌ ์ ์์ต๋๋ค. ๋ฌผ๋ก dict ํ๋ ๋ฐํ์ ๊ฐ๋ฅํ์ง๋ง dict์ฒ๋ผ Wrapper ์๋ฃํ์๋ ๋ฐ๋์ String๋ง ์ฌ์ฉํด์ผ ํฉ๋๋ค. String์ ๋ฐํํ๋ ๋ฉ์๋๋ ํจ์ ๋ํ ์ฌ์ฉํ ์ ์์ต๋๋ค.
๋ฐ๋ผ์ ์๋ฒ๊ฐ ์ ๋๋ก ๋์ํ๋์ง ๊ฐ๋จํ ํ์ธํ ์ ์๋ Health Check์ฉ์ ์ ํฉํฉ๋๋ค.
GET ๋ฉ์๋์์๋ ๋ค์ํ ์๋ฃํ์ ๋ฐํํ ์ ์์ต๋๋ค. JSONResponse ๊ฐ์ฒด๋ฅผ ์ด์ฉํ์ฌ dict ์๋ฃํ์ JSON์ผ๋ก ๋ณํํ ์ ์์ผ๋ฉฐ ์ด ๋๋ json.dumps ๋ฉ์๋๊ฐ ์ฌ์ฉ๋ฉ๋๋ค.
URI์ ๋ณ์๋ฅผ ์ฌ์ฉํ๊ณ ์ ํ๋ ๊ฒฝ์ฐ ์ค๊ดํธ๋ก ๋งคํํด์ฃผ์๋ฉด ๋ฉ๋๋ค.
๊ธฐ๋ณธ API ๋ง๋ค๊ธฐ (POST)
POST ๋ฉ์๋๋ CREATE์ ํด๋นํฉ๋๋ค. ์๋ฒ์ ๋ฐ์ดํฐ๋ฅผ ์ ์ฅํ๊ณ ์ ํ๋ ๊ฒฝ์ฐ ํด๋น ๋ฉ์๋๋ฅผ ์ฌ์ฉํ ์ ์๋๋ฐ์. FastAPI์์๋ ์๋์ ๋ฉ์๋๋ก ์ฌ์ฉํ์ฌ ๊ตฌํํ ์ ์์ต๋๋ค.
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import JSONResponse
app = FastAPI()
class Item(BaseModel):
user_id: str
password: str
@app.post("/register")
async def register_item(item: Item):
dicted_item = dict(item)
dicted_item['success'] = True
return JSONResponse(dicted_item)
FastAPI์์ POST ๋ฉ์๋๋ฅผ ์์ฑํ ๋ ์ฌ์ฉํ๋ ์คํค๋ง ๋ํ๋์๊ฐ ์์ต๋๋ค. ๋ฐ๋ก pydantic์ธ๋ฐ์. ์ด์ ์ ์ ๋ Python์์ ์น ๊ฐ๋ฐํ์ ๋๋ ์คํค๋ง ์ฒดํน์ผ๋ก marshmallow๋ฅผ ๋ง์ด ์ฌ์ฉํ์์ต๋๋ค. ์ต๊ทผ์ Pydantic์ผ๋ก ๊ฐ์ํ๋ฉด์๋ถํฐ ์ฑ๋ฅ ํฅ์์ ๋ง์ด ์ฒดํํ์๋๋ฐ์.
FastAPI์์๋ ์ด๋ฅผ ๊ธฐ๋ณธ์ ์ผ๋ก ์ฑํํ๊ธฐ ๋๋ฌธ์ Pydantic์์ ์ ๊ณตํ๋ BaseModel ์์ ์์ด๋ POST ๋ฉ์๋๋ฅผ ์์ฑํ ์ ์์ต๋๋ค.
BaseModel ์์ ์์ด ๊ตฌํํ ๊ฒฝ์ฐ ์์ ๊ฐ์ด Item ํด๋์ค๊ฐ pydantic field type์ด ์๋๋ผ๋ ์ค๋ฅ๋ฅผ ๋ด๋ฑ์ต๋๋ค.
๊ธฐ๋ณธ API ๋ง๋ค๊ธฐ (PUT, PATCH)
PUT๊ณผ PATCH ๋ฉ์๋๋ UPDATE์ ํด๋นํ๋ ๋ฉ์๋์ ๋๋ค. ๋ ๊ฐ์ ์ฐจ์ด๋ ๋ชจ๋ ๋ด์ฉ์ ๋ฐ๊พธ๋(PUT), ์๋๋ฉด ์ผ๋ถ์ ๋ด์ฉ์ ๋ฐ๊พธ๋(PATCH)์ ์ฐจ์ด์ธ๋ฐ์. ์ด๋ค์์ผ๋ก ๊ตฌํํ ์ ์๋์ง ์์๋ณด๋๋ก ํ๊ฒ ์ต๋๋ค.
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import JSONResponse
app = FastAPI()
class Item(BaseModel):
user_id: str
password: str
@app.post("/register")
async def register_item(item: Item):
global dicted_item
dicted_item = dict(item)
dicted_item['success'] = True
return JSONResponse(dicted_item)
@app.put("/update")
async def update_item(item: Item):
dicted_item = {k:v for k, v in dict(item).items()}
dicted_item['success'] = True
return JSONResponse(dicted_item)
PUT ๋ฉ์๋๋ ์์ ๊ฐ์ด ๊ตฌํํ๋ฉด ์ฝ๊ฒ ๊ตฌํํ ์ ์์ต๋๋ค. register ํฌ์ธํธ๋ฅผ ํตํด ๋ฑ๋กํ ํ์ ์ ๋ณด๋ฅผ update ํฌ์ธํธ๋ฅผ ํธ์ถํด ์์ฒญํ๋ฉด ์ ๋ถ ๋ณ๊ฒฝ๋ฉ๋๋ค.
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import JSONResponse
app = FastAPI()
class Item(BaseModel):
user_id: str
password: str
@app.post("/register")
async def register_item(item: Item):
global dicted_item
dicted_item = dict(item)
dicted_item['success'] = True
return JSONResponse(dicted_item)
@app.put("/update")
async def update_item(item: Item):
dicted_item = {k:v for k, v in dict(item).items()}
dicted_item['success'] = True
return JSONResponse(dicted_item)
@app.patch("/update")
async def update_item_sub(item: Item):
for k, v in dict(item).items():
dicted_item[k] = v
dicted_item['success'] = True
return JSONResponse(dicted_item)
์ ๊ทธ๋ผ PATCH API๋ ์ด๋จ๊น์? ์ด๋ ๊ฒ ๊ตฌํํ๋ฉด ๋ ๊ฒ ๊ฐ์ ๋ณด์ด์ง๋ง...
๋ถํํ๊ฒ๋ ์์ ๊ฐ์ 422 ์ค๋ฅ๋ฅผ ๋ด๋ฟ์ต๋๋ค. ์ ๊ทธ๋ด๊น์? PATCH ๋ฉ์๋๋ ์ผ๋ถ๋ฅผ ๋ณ๊ฒฝํ๋ค๊ณ ํ์ฌ Password๋ง ๋ณ๊ฒฝํ๋ค๋ฉด ์ด๋ฐ์์ผ๋ก ์ค๋ฅ๋ฅผ ๋ํ๋ด๊ฒ ๋ฉ๋๋ค.
์์ธ์ ๋ฐ๋ก Pydantic์ผ๋ก ๊ตฌํํ Item ํด๋์ค์ ์์ต๋๋ค.
class Item(BaseModel):
user_id: str
password: str
ํด๋น ํด๋์ค๋ฅผ ์์ ๊ฐ์ด ๊ตฌํํ๋ฉด ๊ฐ ๋ฉค๋ฒ ๋ณ์๋ Required ํํ๊ฐ ๋ฉ๋๋ค. ๊ทธ๋ ๊ฒ ๋ ๊ฒฝ์ฐ ๋ฐ๋์ ๋ฉค๋ฒ ๋ณ์์ ํด๋นํ๋ ๊ฐ์ด ๋ฐ๋์ ์กด์ฌํด์ผ ํ๋๋ฐ, PATCH ๋ฉ์๋์์๋ user_id๋ฅผ ์ ๋ ฅํ์ง ์์๊ธฐ ๋๋ฌธ์ ์ค๋ฅ๊ฐ ๋ํ๋๋ ๊ฒ์ ๋๋ค.
from typing import Optional
class PatchItem(BaseModel):
user_id: Optional[str]
password: Optional[str]
@app.patch("/update")
async def update_item_sub(item: PatchItem):
dicted_item = {}
for k, v in dict(item).items():
if v:
dicted_item[k] = v
dicted_item['success'] = True
return JSONResponse(dicted_item)
๋ค์ ๋ฒ๊ฑฐ๋กญ์ง๋ง ์ด๋ฐ ๊ฒฝ์ฐ๋ PATCH๋ฅผ ์ํ ํด๋์ค๋ฅผ ํ๋ ๋ ์์ฑํด์ผ ํฉ๋๋ค. ์ฌ๊ธฐ์ ๊ฐ ๋ฉค๋ฒ๋ค์๊ฒ Optional ํ์ ์ ๋ถ์ฌ์ค์ผ๋ก์จ Pydantic์ด ํด๋น ๊ฐ ์ค ์ด๋ค ํ๋๋ผ๋ ์ ๋ ฅ๋์ง ์์ ๊ฒฝ์ฐ์๋ผ๋ API๊ฐ ํ์ฉ๋ ์ ์๋๋ก ๊ตฌํํด์ค์ผ ํฉ๋๋ค.
๊ธฐ๋ณธ API ๋ง๋ค๊ธฐ (DELETE)
์ญ์ API๋ delete ๋ฉ์๋๋ฅผ ์ด์ฉํด์ ์ฝ๊ฒ ๊ตฌํํ ์ ์์ต๋๋ค.
@app.delete("/delete")
async def delete_item():
dicted_item = None
return Response(status_code=HTTP_204_NO_CONTENT)
์ญ์ ๋ฉ์๋ ์ฌ์ฉ์์๋ ๋ณดํต ๋ฐํ ๊ฐ์ผ๋ก EmptyResponse๋ฅผ ์ฌ์ฉํฉ๋๋ค. ๊ทธ๋ฌ๋ Starlette์์๋ EmptyResponse ๊ฐ์ฒด๋ฅผ ๋ณ๋๋ก ์ ๊ณตํด์ฃผ์ง ์์ผ๋ฏ๋ก Response ๊ฐ์ฒด์ 204 ์ฝ๋๋ฅผ ์ฌ์ฉํ์ฌ ๊ตฌํํ ์ ์์ต๋๋ค.
API ๋ฌธ์ ๋ง๋ค๊ณ ๋ค๋ฌ๊ธฐ
์์ ์ฝ๋๋ฅผ ์์ฑํ๊ณ , http://127.0.0.1:8000/redoc ํน์ http://127.0.0.1:8000/docs ๋ก ์ ์ํ๋ฉด ์๋์ ๊ฐ์ด ์ด๋ ์ ๋์ ๋ฌธ์๊ฐ ๋ง๋ค์ด ์ก์์ ์ ์ ์์ต๋๋ค.
์ ํ๋ฉด์ redoc์ ๋ชจ์ต์ ๋๋ค. ์ฐ๋ฆฌ๋ ์คํค๋ง ๋ชจ๋ธ๋ก Pydantic์ ์ฌ์ฉํ์๋๋ฐ์. FastAPI๋ Pydantic์์ ์ ๊ณตํ๋ BaseModel์ ๊ธฐ๋ฐ์ผ๋ก ํ์ฌ๊ธ Request, Response ํ์ ์ ๋ฌธ์๋ฅผ ์๋ํ ์์ผ์ค๋๋ค.
๊ทธ๋ฌ๋ ๋ฌธ์๊ฐ ๊ธฐ๋ณธ์ ์ผ๋ก ํ์ ๋ฑ ๊น๋ํ๊ฒ ๋ณด์ฌ์ฃผ๊ณ ์๊ธด ํ์ง๋ง ์กฐ๊ธ ์ค๋ช ์ด ๋ถ์กฑํ ๋ชจ์ต์ ๋๋ค. ๋ฌธ์์ ์ค๋ช ๋ฑ์ ์ถ๊ฐํ๊ณ , ์ด๋ฅผ ๋ค๋ฌ๊ธฐ ์ํด์ ๋ฉํ๋ฐ์ดํฐ๋ฅผ ์ฌ์ฉํ ์ ์์ต๋๋ค.
@app.get("/{name}", description="์ฌ์ฉ์ ์ด๋ฆ์ ๋ฐ๊ณ ID๋ฅผ ์์ฑํ๋ API ์
๋๋ค.")
async def generate_id_for_name(name: str):
return JSONResponse({
'id': str(uuid.uuid4()),
'name': name
})
๊ฐ decorator์๋ description ํ๋ผ๋ฏธํฐ๋ฅผ ์ ๊ณตํฉ๋๋ค. ์ด ํ๋ผ๋ฏธํฐ์ ๊ฐ๋จํ ์ค๋ช ์ ๋ฃ์ผ๋ฉด...
์ด๋ ๊ฒ API ์ ๋ชฉ ๋ฐ์ ๊ฐ๋จํ ์ค๋ช ์ด ํ์๋ฉ๋๋ค. ๋ง์ฝ ์ ๋ชฉ๋ ๋ฐ๊พธ๊ณ ์ถ๋ค๋ฉด...
@app.get("/{name}", name="์ฌ์ฉ์ ID ์์ฑ", description="์ฌ์ฉ์ ์ด๋ฆ์ ๋ฐ๊ณ ID๋ฅผ ์์ฑํ๋ API ์
๋๋ค.")
async def generate_id_for_name(name: str):
return JSONResponse({
'id': str(uuid.uuid4()),
'name': name
})
name ํ๋ผ๋ฏธํฐ๋ฅผ ์ฌ์ฉํ์ฌ ์ฝ๊ฒ ๋ฐ๊ฟ ์ ์์ต๋๋ค.
JSON ๋ฐ์ดํฐ๋ฅผ ์ ๋ ฅํ๋ ๋ฉ์๋์ ๊ฒฝ์ฐ ์์ฒญ ํ๋ผ๋ฏธํฐ๋ ๋์ ์์ง๋ง ์๋ต ํ๋ผ๋ฏธํฐ์ ๋ํ ์์ธ ์ฌํญ์ ๋์ ์์ง ์์ต๋๋ค. ๊ทธ ์ด์ ๋ ์๋ต ๋ชจ๋ธ์ ๋ํ ํด๋์ค๋ฅผ ๊ตฌํํ์ง ์์๊ธฐ ๋๋ฌธ์ ๋๋ค.
class Item(BaseModel):
user_id: str
password: str
class ResponseItem(Item):
success: bool
@app.post("/register", response_model=ResponseItem)
async def register_item(item: Item):
global dicted_item
dicted_item = dict(item)
dicted_item['success'] = True
return JSONResponse(dicted_item)
์์ฒ๋ผ Response Model์ ์ ์ํ๋ฉด ๋ฌธ์์ ์ํ์ด ํ์๋ฉ๋๋ค. ๊ทธ๋ฌ๋ ์ฌ์ ํ ๋ถ์กฑํ ์ ์ด ์์ต๋๋ค. ๊ฐ ํ๋ผ๋ฏธํฐ๊ฐ ์ด๋ค ๊ฑธ ๋ํ๋ด๋์ง ํ ๋จ์ด๋ก ํ์ํ๊ธฐ ์ด๋ ค์ธ ๊ฒฝ์ฐ ๊ฐ๋จํ description์ด ์๋ค๋ฉด ์ข์ ๊ฒ ๊ฐ์ต๋๋ค.
class Item(BaseModel):
user_id: str = Field(title='์ฌ์ฉ์๊ฐ ์ฌ์ฉํ ID')
password: str = Field(title='์ฌ์ฉ์๊ฐ ์ฌ์ฉํ Password')
class ResponseItem(Item):
success: bool = Field(True, "์ฒ๋ฆฌ ์ฌ๋ถ/๊ฒฐ๊ณผ")
class PatchItem(BaseModel):
user_id: Optional[str] = Field(None, title='์ฌ์ฉ์๊ฐ ์ฌ์ฉํ ID')
password: Optional[str] = Field(None, title='์ฌ์ฉ์๊ฐ ์ฌ์ฉํ Password')
์ด๋ฐ ๊ฒฝ์ฐ์๋ Pydantic์์ ์ ๊ณตํ๋ Field๋ฅผ ์ฌ์ฉํด๋ณผ ์ ์์ต๋๋ค. ์ด๋ ๊ฒ ํ๋ฉด ํ์ ๊ณผ ํจ๊ป ๊ฐ๋จํ ์ฝ๋ฉํธ๊ฐ ํ์๋จ์ผ๋ก์จ ํ๋ก ํธ์๋ ๊ฐ๋ฐ์๊ฐ ์ข ๋ ์์ธํ๊ฒ API ๋ฌธ์๋ฅผ ๋ณผ ์ ์๋ ์ฅ์ ์ ์ ๊ณตํฉ๋๋ค.
๋ง์น๋ฉฐ...
์ฌ๊ธฐ๊น์ง ์์ฃผ ๊ฐ๋จํ API๋ฅผ ๋ง๋ค์ด๋ณด๋ฉด์ ๋ฌธ์ ์๋ํ ๊ธฐ๋ฅ์ ๋ํ ๊ฒ์ ๋ํด ์์๋ดค์ต๋๋ค. Pydantic์ ์ฑ๋ฅ์ ๊ฐ์ ธ๊ฐ๊ณผ ๋์์ FastAPI๊ฐ ์ ๊ณตํ๋ API ๋ฌธ์ ์๋ํ๋ REST API ์๋ฒ๋ฅผ ์ฝ๊ณ ์ฌ๋ฐ๊ฒ ๊ฐ๋ฐํด์ฃผ๋ ํ๋ฌ์ค ์์๊ฐ ๋์์ต๋๋ค.
๋์ฑ์ด ๊ฐ๋จํ ๋ฌธ๋ฒ์ ํตํ์ฌ ๋ฒ๊ทธ์ ์ค๋ฅ๋ฅผ ์ค์ด๋ ์ญํ ์ ํ๋ ๊ฒ์ ์์ฃผ ์ข์ ์๋์ง๋ฅผ ๋ณด์ฌ์ฃผ์๊ณ , ๋น๋๊ธฐ ์ฒ๋ฆฌ๋ฅผ ์ ๊ณตํจ์ผ๋ก์จ ํ์ด์ฌ์์ ์น ํ๋ ์์ํฌ๋ฅผ ๋ณด๋ค ์ฝ๊ณ ๊ฐ๋ ฅํ๊ฒ ๊ฐ๋ฐํ ์ ์์๋ค๊ณ ์๊ฐํ๋๋ฐ์.
๋ค์ ๊ธ ๋ถํฐ๋ ๋ฐ์ดํฐ๋ฒ ์ด์ค์ ์ฐ๋์ ์์์ผ๋ก ORM์ ์ด์ฉํ FastAPI ํ๋ก๊ทธ๋๋ฐ์ ๋ํด ์ข ๋ ์ฌ์ธต์ ์ผ๋ก ๋ค๋ค๋ณด๋๋ก ํ๊ฒ ์ต๋๋ค.