Fast API, Request Body (POST, PUT) 받기

Junha Kim·2021년 1월 3일
4

FastAPI

목록 보기
7/16
post-custom-banner

지금까지는 GET방식이기에 URL 및 Request Header로 값이 넘어왔다. 하지만 POST방식에서는 Request Body로 넘어오기에 이전가지의 방식으로는 데이터를 받을 수 없다.

Request Body는 Pydantic Model을 이용하여 값을 받을 수 있다.


Pydantic BaseModel

먼저 Pydantic모듈의 BaseModel을 import하여 Class를 구성한다.

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
		pass

이후 Item클래스에서 Requsest Body로 받을 데이터를 정의하면 된다.

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

이전에 parameter를 정의한 것과 같이, 형식과 Default 값을 명시해준다.

위와 같이 선언이 되면 description, tax는 optional value가 되며, 나머지는 required value가 된다.

{
    "name": "Foo",
    "price": 45.2
},
{
    "name": "Foo",
    "description": "An optional description",
    "price": 45.2,
    "tax": 3.5
}

POST Method Parameter

이제 path를 생성하고 함수에서 어떻게 데이터를 받는지 볼 차례이다.

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return item

create_item함수에서 item이라는 parameter를 받는다. 이때의 형식은 위에서 우리가 정의한 Item 객체라고 명시한다.

위와 같이 정의되었을 때, FastAPI에서 처리하는 과정은 아래와 같다.

  1. JSON으로 넘어온 Request Body 내용을 읽는다.
  2. 필요시 해당 데이터를 상응되는 타입으로 변환한다.
  3. 데이터 유효성 검사를 한다. → 만약 invalid하면 error를 반환한다.
  4. 받은 데이터를 파라미터 item에 넘겨준다.
  5. Item 객체에 맞게 JSON Schema를 생성하게 되고, 함수 내부에서 item은 Item 객체로서 사용된다.

사전에 정의되지 않은 객체 변수를 추가할 수 있다.

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

Request Body + Path Parameters

Request Body와 Path Parameter를 동시에 사용도 가능하다.

FastAPI는 Path parameter는 path에서 찾고, Pydantic model로 선언된 parameter는 Request Body에서 찾게 된다.

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.dict()}

Request Body + Path Parameters + Query Parameters

Query parameter 또한 함께 사용 가능하다

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Optional[str] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result
post-custom-banner

0개의 댓글