[FastAPI] Tutorial(2)

728x90

※ FastAPI Tutorial을 공부하며 이해한대로 작성한 내용으로, 실제 내용과 다르거나 틀린 부분이 있을 수 있습니다. 댓글로 알려주시면 즉시 반영하겠습니다. 감사합니다.

https://fastapi.tiangolo.com/ko/tutorial/

[자습서 - 사용자 안내서 - 도입부 - FastAPI

자습서 - 사용자 안내서 - 도입부 이 자습서는 FastAPI의 대부분의 기능을 단계별로 사용하는 방법을 보여줍니다. 각 섹션은 이전 섹션을 기반해서 점진적으로 만들어 졌지만, 주제에 따라 다르게

fastapi.tiangolo.com](https://fastapi.tiangolo.com/ko/tutorial/)

1. Body - 여러 개의 매개변수

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

@app.put("/items/{item_id}")
async def update_item(
    item_id: Annotated[int, Path(title="the ID of the item to get", ge=0, le=1000)],
    q: Union[str, None] = None,
    item: Union[Item, None] = None
):
    result = {"item_id": item_id}
    if q:
        result.update({"q": q})
    if item:
        result.update({"item": item})

    return result

Body - 단일 변수

@app.put("/body/{item_id}")
async def update_body(
    item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

importance: Annotated[int, Body()] ⇒ 이러면 데이터 목록-그러니까 response body에 포함됨
이러지 않으면 Query 매개변수로 판단되고 데이터 바디로 들어가지 않음. 위를 실행하면 아래처럼 나온다.
- query 매개변수는 주소창에 표시되고, body에 포함되는 친구들은 표시되지 않는다.
- 보안상의 의미가 큰 것 같다.
- { "item_id": 3, "item": { "name": "hell", "description": "biohazard", "price": 0, "tax": 0 }, "user": { "username": "mia", "full_name": "edan" }, "importance": 10 }

다중 변수

@app.put("/body/{item_id}")
async def update_body(
    item_id: int, item: Item, user: User, importance: Annotated[int, Body()], q: Union[str, None] = None
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

위와 같이 일반 쿼리 매개변수와 함께 쓰는 것도 당연히 가능함

딱 하나만 body에 포함시키고자 할 때

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results

Body(embed=True) 를 붙이면 왼쪽이 오른쪽처럼 변함

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

2. Body - Field

⇒ pydantic model 안에서 속성을 추가할 때 = field

from typing import Union

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field
from typing_extensions import Annotated

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Union[str, None] = Field(
        default=None, title="The description of the item", max_length=300
    )
    price: float = Field(gt=0, description="The price must be greater than zero")
    tax: Union[float, None] = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results

3. Body - Nested models

from typing import List, Set, Union

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()

class Image(BaseModel):
    url: HttpUrl
    name: str

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
    tags: Set[str] = set()
    images: Union[List[Image], None] = None

class Offer(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    items: List[Item]

@app.post("/offers/")
async def create_offer(offer: Offer):
    return offer

연쇄적으로 정의한 클래스를 자료형마냥 사용하는 예제
단순 리스트로 구성된 body도 사용 가능

from typing import List

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Image(BaseModel):
    url: HttpUrl
    name: str


@app.post("/images/multiple/")
async def create_multiple_images(images: List[Image]):
    return images

dictionary의 경우 key와 value의 자료형을 따로.

@app.post("/index-weights/")
async def create_index_weights(weights: Dict[int, float]):
    return weights

4.Request 예시 데이터 선언

⇒ Pydantic model을 기반으로 한 클래스에서는 config라는 서브 클래스와 schema_extra라는 거 쓰면 된다.

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

    class Config:
        schema_extra = {
            "example": {
                "name": "Foo",
                "description": "A very nice Item",
                "price": 35.4,
                "tax": 3.2,
            }
        }

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

# 기본 예시데이터가 request body에 미리 있음. JSON
{
  "name": "Foo",
  "description": "A very nice Item",
  "price": 35.4,
  "tax": 3.2
}

Field 에도 따로 example을 넣어주는 방법도 있다.

Body로 Annotated할 때 예시 넣기

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[
        Item, 
        Body(
            example={
                "name": "Foo",
                "description": "A very nice item",
                "price": 24.3,
                "tax": 3.3
            }
        )
    ]
):
    results = {"item_id": item_id, "item": item}
    return results

여러 개의 example 넣기?

@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int,
    item: Annotated[
        Item,
        Body(
            examples={
                "normal": {
                    "summary": "A normal example",
                    "description": "A **normal** item works correctly.",
                    "value": {
                        "name": "Foo",
                        "description": "A very nice Item",
                        "price": 35.4,
                        "tax": 3.2,
                    },
                },
                "converted": {
                    "summary": "An example with converted data",
                    "description": "FastAPI can convert price `strings` to actual `numbers` automatically",
                    "value": {
                        "name": "Bar",
                        "price": "35.4", # string to number automatically
                    },
                },
                "invalid": {
                    "summary": "Invalid data is rejected with an error",
                    "value": {
                        "name": "Baz",
                        "price": "thirty five point four", # float인데 string
                    },
                },
            },
        ),
    ],
):
    results = {"item_id": item_id, "item": item}
    return results

5. Cookie 매개변수

from typing import Union

from fastapi import Cookie, FastAPI
from typing_extensions import Annotated

app = FastAPI()

@app.get("/items/")
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
    return {"ads_id": ads_id}

6. Header 매개변수

메타데이터에 대한 정보를 담고 있음.

from typing import Union

from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: Union[str, None] = Header(default=None)):
    return {"User-Agent": user_agent}

→ 기본적으로 path, query, cookie와 같은 기능

추가 기능 : 언더바(_)를 하이픈(-)으로 자동 변환 HTTP 표준 헤더 규격
중복적으로 header를 전송하는 경우 파이썬 리스트로 저장

7. 응답 모델 - 반환형

1. 일반 파이썬 docstring 문법대로 함수의 반환형을 표기하기

@app.post("/items/")
async def create_item(item: Item) -> Item: # return type = Item object
    return item

@app.get("/items/")
async def read_items() -> List[Item]: # return type = list of Item object
    return [
        Item(name="Portal Gun", price=42.0),
        Item(name="Plumbus", price=32.0),
    ]

2. `response_model` parameter 사용하기

유사시 파이썬 오류를 막기 위해서 쓸 수 있음. FastAPI에서 출력 데이터를 response model의 자료형으로 변환 및 필터링

@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:
    return item

@app.get("/items/", response_model=List[Item])
async def read_items() -> Any:
    return [
        {"name": "Portal Gun", "price": 42.0},
        {"name": "Plumbus", "price": 32.0},
    ]

따라서 입력받은 데이터 중 일부를 필터링해서 출력으로 돌리고 싶으면 그것도 가능!

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user

이 경우 입력 데이터에서 password를 필터링한 UserOut으로 자동 변환
python 일반 함수 반환에다가 쓰면 둘이 달라서 안된다고 오류남.

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class BaseUser(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None


class UserIn(BaseUser): # BaseUser 상속!
    password: str
    # username: str
    # email: EmailStr
    # full_name: Union[str, None] = None

@app.post("/user/")
async def create_user(user: UserIn) -> BaseUser:
    return user

⇒ 이 경우 BaseUser에 UserIn이 가진 인스턴스가 하나도 없는데?

상속 받았으니까 UserIn에 BaseUser가 가진 것이 상속되어있음. 따라서 상속받은 것들은 필터에 안 걸리고 없는 password만 걸림.
결과

# request body
{
  "username": "dain",
  "email": "dmelli0505@gmail.com",
  "full_name": "dain kang",
  "password": "123456789"
}

# response body
{
  "username": "dain",
  "email": "dmelli0505@gmail.com",
  "full_name": "dain kang"
}

3. 응답 직접 반환

from fastapi.responses import JSONResponse, RedirectResponse
from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/portal")
async def get_potal(teleport: bool = False) -> Response:
    if teleport:
        return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
    return JSONResponse(content={"message": "Here's your interdimenseional portal."})

더 하위 response를 반환하는 것도 ok.

그러나 반환형이 A or B는 안됨.

@app.get("/portal")
async def get_portal(teleport: bool = False) -> Union[Response, dict]:
    if teleport:
        return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
    return {"message": "Here's your interdimensional portal."}

이 경우 Response or Dictionary 반환? 명시적이지 않으므로 불가능
@app.get("/portal/", response_model=None) 하면 비활성화

4. 응답 모델 매개변수

⇒ 응답모델도 기본값을 가질 수 있음.

class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10.5
    tags: List[str] = []

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)

이렇게 쓰면 기본값 있는 것들 빼고 나머지만 response로 들어감.
- { name: "temp", price: 2134 }
그러나 미리 데이터에 기본값과 같은 값이 들어있는 경우에는 평범하게 전부 response로 들어감.

items = [
	{
		name: "temp",
		description: None,
		price: 2134,
		tax: 10.5,
		tags: []
	},
	...
]

# temp.com/items/0 -> response
	{
		name: "temp",
		description: None,
		price: 2134,
		tax: 10.5,
		tags: []
	}

8. 입력 / 출력 / DB 모델

입력 모델 = 비밀번호 필요
출력 모델 = 비밀번호 포함하면 안됨
DB 모델 = 해시 비밀번호가 필요할 수 있음

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None

class UserInDB(BaseModel):
    username: str
    hashed_password: str
    email: EmailStr
    full_name: Union[str, None] = None

→ 유저 데이터를 가공하는 부분

def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ...really?")
    return user_in_db

return type UserInDB
**user_in.dict() 는 Pydantic model의 method인데, model data에 대한 dictionary를 제공한다.

**user_in.dict() ⇒ {
  "username": "dump",
  "password": "1234",
  "email": "dump@example.com",
  "full_name": "dump dumpio"
}

그리고 겹치는 key에만 unwrapping 하므로 UserInDB 에서는 username, email, full_name만 남게 된다. 거기에 hashed_password를 더해서 구성한 것.

@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in=user_in)
    return user_saved

결과

# Request body
{
  "username": "dump",
  "password": "1234",
  "email": "dump@example.com",
  "full_name": "dump dumpio"
}

# Response body
{
  "username": "dump",
  "email": "dump@example.com",
  "full_name": "dump dumpio"
}

이중 선언 지양하기

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None

class UserIn(UserBase):
    password: str

class UserOut(UserBase):
    pass

class UserInDB(UserBase):
    hashed_password: str

상속받아서 길이 줄이기

기타 response model

리턴 모델이 둘 중 하나일 때 ⇒ Union

@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])  # Union으로
async def read_item(item_id: str):
    return items[item_id]

리턴 모델을 List 로 사용도 가능

@app.get("/items/", response_model=List[Item])
async def read_items():
    return items

field / attribute 이름알 잘 모를 때 Dictionary 형태로도 response를 반환 가능하다.

from typing import Dict

from fastapi import FastAPI

app = FastAPI()


@app.get("/keyword-weights/", response_model=Dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

Dict[str, float] 이므로 key = string, value = float 형인 딕셔너리가 return 될 예정임을 명시

9. 응답 상태 코드

@app.get()
@app.post()
@app.put()
@app.delete()

와 같은 동작을 수행했을 때 404 error와 같이 응답 상태 코드를 받아볼 수 있는데, 이를 status_code 매개변수로 지정할 수 있다.

from fastapi import FastAPI

app = FastAPI()

@app.post('/items/', status_code=201)
async def create_item(name:str):
    return {"name": name}

HTTP 상태코드

1xx : 정보. 직접적으로 사용되지 않으며 body를 가질 수 없음
2xx : 성공적인 응답.
- 200 : 전체 성공
- 201 : 생성됨 - DB에 새로운 레코드 생성 후 사용
- 204 : 내용 없음 - 클라이언트에 반환할 것 없음. body를 가질 수 없음.
3xx : 리다이렉션용 - body가 있을 수도 있고, 없을 수도 있음.
- 304 : 수정되지 않음 - body 없음
4xx : 클라이언트 오류
- 404 : 찾을 수 없음
- 400 : 일반 클라이언트 오류
5xx : 서버 오류

FastAPI에서 편의 변수 사용

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}