[FastAPI] Docs Bookmark - Tutorial 2

python 3.6 ~ python 3.8

Fastapi 0.95.0

---

Cookie Parameters

⌾ import Cookie

Cookie로 선언하지 않으면, 쿼리 파라미터로 인식된다

from typing import Annotated
from fastapi import Cookie, FastAPI
...
@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
	return {"ads_id": ads_id}

---

Header Parameters

⌾ import Header

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

⌾ 자동 변환 

⌵ 대부분의 표준 header는 하이픈 문자 '-'로 구분되나, 이는 snake_case로 알려진 표준 파이썬 스타일과 맞지 않다

⌵ 따라서 Header는 기본적으로 언더스코어 '_'에서 하이픈 '-'로 변환하여 header를 추출한다

⌵ 또한, HTTP header는 대소문자를 구분하지 않으므로 첫 문자를 대문자화할 필요도 없다

# 만약 하이픈 자동 변환을 비활성화하고 싶다면: convert_underscores=False
...
async def read_items(
	strange_header: Union[str, None] = Header(default=None, convert_underscores=False)
):

⌾ 중복 헤더

다중값을 갖는 동일 헤더는 list를 사용한다

...
async def read_items(x_token: Union[List[str], None] = Header(default=None)):
	return {"X-Token values": x_token}

---

Response Model - Return type

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

⌾ response_model 

⌵ 손쉽게 output 데이터를 정형화하여 내보낼 수 있다 

⌵ decorator method (get, post, ...)의 파라미터

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

# return 값이 Item에 정의된 형식에 맞춰서 내보내짐
@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:
    return item

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

⌾ Return the same input data & Add an output model

...
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr

class UserOut(BaseModel):
    username: str
    email: EmailStr
    
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user

 Input에 UserIn 형식, response_model에 UserOut을 선언함으로써 password 값은 내보내지 않는다.

• UserIn : 새로운 유저를 만들기 위한 모델
• UserOut : 생성된 유저에 대한 정보를 다시 클라이언트에게 보내는 모델

⌾ Return type & Data Filtering

...
class BaseUser(BaseModel):
    username: str
    email: EmailStr

class UserIn(BaseUser):
    password: str

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

위 두 코드 모두 똑같은 input/output 형식을 가진다

Other Return Type Annotations

⌾ Return a Response Directly

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

app = FastAPI()

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

⌾ Annotate a Response Subclass

⌾ Invalid Return Type Annotations ⎐ Disable Response Model 

response_model=None

from typing import Union
from fastapi import FastAPI, Response
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/portal", response_model=None)
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_model_exclude_unset 

response model이 default 값을 가질 때, request로 덮어씌워진 값만 내보내는 방법
(default 값들중에서 바뀌지 않은 값들은 제외시키되, default 값과 똑같은 값은 새로씌워진 것으로 판단하여 반환된다.)

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

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

 

위 경우에서, ID foo에 대한 응답은 다음과 같다. 

{"name": "Foo", "price": 50.2}

• response_model_exclude_defaults=True : default 값과 똑같은 값일 경우, 해당 필드 반환하지 않음
• response_model_exclude_none=True : None 값을 가진 필드를 반환하지 않음

Exporting models - Pydantic

⌾ Response Model Include/Exclude

• 작성되어 있는 pydantic 모델에서 특정 필드만 내보내고 싶거나, 특정 필드를 제외하고 나머지 필드를 내보내고 싶을 때 사용
• set(str) 사용
  list(str), tuple(str)을 사용하는 경우 FastAPI가 이를 집합으로 변환하여 작동시킨다

...
@app.get(
    "/items/{item_id}/name",
    response_model=Item,
    response_model_include={"name", "description"}, #set
)
async def read_item_name(item_id: str):
    return items[item_id]

@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude={"tax"})
async def read_item_public_data(item_id: str):
    return items[item_id]

---

Extra Models

user model에서,

• input 모델은 password가 있어야 한다
• output 모델은 password가 없어야 한다
• database 모델은 해시된 password가 있어야 한다

user model을 Multiple model로 구현했을 때, 중복된 내용이 많음을 볼 수 있다.

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


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! ..not really")
    return user_in_db


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

About **user_in.dict()

pydantic 모델에는 데이터를 dict 형식으로 반환하는 .dict() 방법이 있다.

pydantic 객체 user_in을 출력해보면,

user_in = UserIn(username="john", password="secret", email="john.doe@example.com")
user_dict = user_in.dict()
print(user_dict)

## result
{
    'username': 'john',
    'password': 'secret',
    'email': 'john.doe@example.com',
    'full_name': None,
}

 이러한 dict를 **user_in.dict() 형태로 전달하면, python이 unwrap 과정을 거쳐 key-value 인수로 값을 직접 전달한다

⌾ Reduce duplication

모델을 상속하는 서브 클래스 작성을 통해 중복 제거

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

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


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! ..not really")
    return user_in_db
    
@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved

---

Response Status Code

response에 사용될 HTTP status code를 숫자로 지정한다

(또는 python의 http.HTTPstatus와 같은 IntEnum을 입력받을 수 있다. (python-http.HTTPStatus)

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

⌾ fastapi.status

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

---

Form Data

Json 대신 form field를 수신해야할 경우

pip install python-multipart

⌾ Import Form

from fastapi import FastAPI, Form
from typing_extensions import Annotated

app = FastAPI()

@app.post("/login/")
async def login(username: Annotated[str, Form()], password: Annotated[str, Form()]):
    return {"username": username}

About Form Fields

HTML <form> </form>에서 데이터를 서버로 전송하는 방식은 JSON과 달리 special encoding을 사용한다

• 일반적으로 forms 데이터는 파일이 포함되지 않은 경우,
  "media type" application/x-www-form-urlencoded를 사용하여 인코딩된다
• 파일이 포함된 경우, multipart/form-data로 인코딩된다
• 다수의 File과 Form 파라미터를 한 path operation에 선언하는 것은 가능하지만,
  json으로 받아야하는 Body 필드를 같이 선언할 수는 없다. 

POST - HTTP | MDN

---

Request Files

⌾ File을 사용하여 클라이언트가 업로드할 파일을 정의할 수 있다 

pip install python-multipart

• File은 Form 데이터의 형태로 업로드 된다
• 전체 내용이 메모리에 저장되므로 작은 크기의 파일들에 적합하다
• 다음과 같이 매개변수를 bytes로 선언하는 경우, FastAPI는 bytes 형태의 내용을 전달한다

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/files/")
async def create_file(file: bytes = File()):
    return {"file_size": len(file)}

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}

⌾ UploadFile

• 'spooled' file을 사용한다
   (최대 크기 제한까지만 메모리에 저장되며, 이를 초과하는 경우 디스크에 저장된다)
   (이미지, 동영상, 큰 이진코드와 같은 대용량 파일들을 많은 메모리를 소모하지 않으면서 처리할 수 있다)
• 업로드 된 파일의 메타데이터를 얻을 수 있다
• async interface를 가지고 있다 (file-like object)
• file-like object를 필요로하는 다른 라이브러리에 직접적으로 전달할 수 있는 파이썬 객체를 반환한다
  (python-SpooledTemporaryFile)

UploadFile Attributes:

• filename : 문자열(str), 업로드된 파일의 파일명 
• content_type : 문자열(str), 파일 형식(MIME type/media type) 
• file : SpooledTemporaryFile
• async methods
  - 내부적인 SpooledTemporaryFile을 사용하여 해당하는 파일 메소드 호출
  - async method이므로 await 사용
  - 일반적인 def 함수라면, Uploadfile.file.read()와 같이 직접 접근
  
  • write(data) : data를 파일에 작성 (str or bytes)
  • read(size) : 파일의 바이트 및 문자의 size(int) 
  • seek(offset) : 파일 내 offset(int)위치의 바이트로 이동
    - await myfile.seek(0)을 사용하면 파일의 시작부분으로 이동
    - await myfile.read()를 사용한 후 내용을 다시 읽을 때 유용
  • close() : 파일 닫기

⌾ 다중 파일 업로드

...
@app.post("/files/")
async def create_files(files: List[bytes] = File()):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile]):
    return {"filenames": [file.filename for file in files]}

---

Request  Forms and Files

pip install python-multipart

⌾ File / Form 파라미터 정의

하나의 요청으로 데이터와 파일들을 받아야 할 경우, File과 Form 함께 사용

from fastapi import FastAPI, File, Form, UploadFile
from typing_extensions import Annotated

app = FastAPI()

@app.post("/files/")
async def create_file(
    file: Annotated[bytes, File()], #non-Annotated file:bytes=File()
    fileb: Annotated[UploadFile, File()], #non-Annotated fileb:UploadFile=File()
    token: Annotated[str, Form()], #non-Annotated token:str=Form()
):
    return {
        "file_size": len(file),
        "token": token,
        "fileb_content_type": fileb.content_type,
    }

---

Handling Errors

⌾ 오류 HTTP 응답을 위한 HTTPException

Add custom headers

from fastapi import FastAPI, HTTPException
...
@app.get("/items/{item_id}")
async def read_item(item_id: str):
    # 클라이언트가 존재하지 않는 id를 요청하면, 404 예외 발생
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"},
        )
    return {"item": items[item_id]}

⌾ Install custom exception handlers

@app.exception_handler()를 이용해 사용자 지정 UnicornException 예외 생성

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name
...
@app.exception_handeler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."}
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

⌾ Override the default exception handlers

⌾ Override the HTTPException error handler

⌾ Use the RequestValidationError body

Error 응답과 함께 invalid data를 사용자에게 반환

from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )

class Item(BaseModel):
    title: str
    size: int

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

⌾ Re-use FastAPI's exception handlers

---

Path Operation Configuration

⌾ Tags & Enum

from enum import Enum
from fastapi import FastAPI

app = FastAPI()

class Tags(Enum):
    items = "items"
    users = "users"

@app.get("/items/", tags=[Tags.items])
async def get_items():
    return ["Portal gun", "Plumbus"]

@app.get("/users/", tags=[Tags.users])
async def read_users():
    return ["Rick", "Morty"]

Tags & Enum

⌾ Description from docstring & Response description

@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    description="Description",
    response_description="The created item",
)
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item

⌾ Deprecate a Path operation

...
@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
    return [{"item_id": "Foo"}]

---

JSON Compatible Encoder

pydantic 모델과 같은 데이터 유형을 JSON과 호환된 형태로 반환해야 하는 경우

(예: 데이터베이스에 저장하는 경우)

⌾ jsonable_encoder

⌵ datetime 객체는 json과 호환되지 않기때문에 ISO format 데이터를 포함하는 str로 변환한다

⌵ 또다른 예로 데이터베이스가 pydantic 모델을 받지 않고, dict만을 받을 때 변환을 위해 사용한다

...
from fastapi.encoders import jsonable_encoder

fake_db = {}

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Union[str, None] = None

app = FastAPI()

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data

호출한 결과는 json.dumps로 인코딩 할 수 있다

---

Body - Updates

⌾ Update replacing with PUT (HTTP-PUT)

PUT은 기존 데이터를 대체하는 데이터를 수신하는 데에 사용된다

...
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_encoded = jsonable_encoder(item)
    items[item_id] = update_item_encoded
    return update_item_encoded

⌾ Partial updates with PATCH (HTTP-PATCH)

PATCH를 사용하여 데이터를 부분적으로 업데이트 할 수 있다

• exclude_unset=True : default 값은 생략하고, 요청된 데이터만 dict 생성
• .copy() : 기존 모델의 복사본에 업데이트 데이터가 담긴 dict로 파라미터 전달

...
## partial updates using PATCH
@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item): 
    stored_item_data = items[item_id] # 저장된 데이터 검색
    stored_item_model = Item(**stored_item_data) # pydantic 모델에 검색한 데이터 삽입 
    update_data = item.dict(exclude_unset=True) # default 값 없이 update dict 생성
    updated_item = stored_item_model.copy(update=update_data) # 모델 복사본에 데이터 update
    items[item_id] = jsonable_encoder(updated_item) # 복사본을 DB에 저장할 수 있도록 변환 후 저장
    return updated_item # 업데이트된 모델 반환

---

Reference

FastAPI

Fast API, Response Model (Output 데이터 형식)