- Handling Errors
- 경로 작동 설정
- JSON 호환 가능 인코더
- Body - Updates
- 미들웨어
- 교차 출처 리소스 공유
API를 개발하다 보면 클라이언트가 API에 접근할 때 발생하는 다양한 오류를 처리해야 합니다. 이러한 오류들은 클라이언트의 요청이 잘못되었거나, 리소스에 접근 권한이 없을 때 발생할 수 있습니다. FastAPI에서는 이러한 오류를 처리하는 방법을 쉽게 제공합니다.
FastAPI는 오류 응답을 쉽게 반환하기 위해 HTTPException을 제공합니다. 이는 일반적인 파이썬 예외처럼 동작하면서, 추가적인 HTTP 관련 데이터를 포함할 수 있습니다.
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Wrestlers"}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
이 코드에서는 item_id가 items 딕셔너리에 없을 경우, 404 Not Found 오류가 발생하게 됩니다.
HTTPException을 사용하면 상태 코드와 함께 오류 메시지를 보낼 수 있습니다. FastAPI는 해당 예외가 발생하면 즉시 요청 처리를 중단하고, 클라이언트에 오류를 반환합니다.
경우에 따라 오류 응답에 커스텀 헤더를 추가하고 싶을 때가 있습니다. FastAPI는 HTTPException에 headers 매개변수를 사용해 추가적인 정보를 제공할 수 있습니다.
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Wrestlers"}
@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
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]}
여기에서는 추가적인 헤더인 X-Error를 반환하여 클라이언트에 오류 정보를 제공합니다.
FastAPI는 사용자 정의 예외를 처리할 수 있는 기능을 제공합니다. 예를 들어, 아래처럼 UnicornException이라는 예외를 정의하고 이를 처리할 수 있습니다.
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class UnicornException(Exception):
def __init__(self, name: str):
self.name = name
app = FastAPI()
@app.exception_handler(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}
이 예제에서는 UnicornException이 발생하면 418 상태 코드와 함께 커스텀 메시지가 반환됩니다.
FastAPI는 기본적으로 RequestValidationError와 같은 예외에 대한 기본 핸들러를 제공합니다. 그러나 이러한 기본 핸들러를 재정의하여 커스텀 처리 로직을 추가할 수 있습니다.
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return PlainTextResponse(str(exc), status_code=400)
위 코드에서는 RequestValidationError가 발생할 때 기본적으로 JSON 대신 단순한 텍스트 응답을 반환하도록 재정의했습니다.
요청 데이터가 잘못되었을 때 FastAPI는 자동으로 RequestValidationError를 발생시키며, 기본적으로 이 오류는 클라이언트에게 422 Unprocessable Entity 응답으로 전달됩니다. 이 기본 동작도 재정의할 수 있습니다.
from fastapi import FastAPI, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
)
FastAPI의 HTTPException은 Starlette의 HTTPException을 상속받습니다. 다만, FastAPI의 HTTPException은 detail 필드에 JSON 형식으로 변환 가능한 모든 데이터를 허용하는 반면, Starlette의 HTTPException은 문자열만 허용합니다.
FastAPI는 기본 제공 예외 핸들러를 재사용할 수 있는 기능을 제공합니다. 이는 코드 중복을 줄이고, 표준 핸들러를 유지하는 데 유용합니다.
from fastapi import FastAPI
from fastapi.exception_handlers import http_exception_handler, request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
app = FastAPI()
@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
print(f"OMG! An HTTP error!: {repr(exc)}")
return await http_exception_handler(request, exc)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
print(f"OMG! The client sent invalid data!: {exc}")
return await request_validation_exception_handler(request, exc)
FastAPI를 사용하면 HTTPException을 통해 오류 응답을 쉽게 처리할 수 있으며, 사용자 정의 예외 핸들러를 추가해 복잡한 오류 처리 로직을 구현할 수 있습니다. 기본 제공되는 예외 핸들러는 필요에 따라 재정의하거나 재사용할 수 있으며, 개발자는 이를 통해 오류 처리 로직을 더욱 세밀하게 제어할 수 있습니다.
경로 작동을 설정하고 더 나은 API 문서를 만들기 위해 다양한 매개변수를 사용할 수 있습니다. 경로 작동 데코레이터에서 제공되는 여러 매개변수를 통해 경로 작동에 대한 메타데이터를 추가하거나 동작을 조정할 수 있습니다. 아래는 주요 매개변수와 그 사용 방법입니다.
status_code 매개변수를 사용하여 경로 작동에서 반환할 HTTP 상태 코드를 정의할 수 있습니다. 상태 코드를 숫자로 지정할 수도 있지만, status 모듈의 상수를 사용하는 것이 더 직관적입니다.
from fastapi import FastAPI, status
from pydantic import BaseModel
from typing import Union, Set
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
tags: Set[str] = set()
@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
return item
status.HTTP_201_CREATED는 201 상태 코드에 대한 상수이며, OpenAPI 스키마에 자동으로 반영됩니다.
tags 매개변수를 사용하여 경로 작동에 태그를 추가할 수 있습니다. 태그는 경로 작동을 그룹화하고, API 문서에서 카테고리를 구분하는 데 유용합니다.
@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):
return item
@app.get("/items/", tags=["items"])
async def read_items():
return [{"name": "Foo", "price": 42}]
@app.get("/users/", tags=["users"])
async def read_users():
return [{"username": "johndoe"}]
경로 작동에 대한 summary와 description을 추가하여 API 문서를 더 읽기 쉽게 만들 수 있습니다.
@app.post(
"/items/",
response_model=Item,
summary="Create an item",
description="Create an item with all the necessary details such as name, description, price, tax, and a set of unique tags",
)
async def create_item(item: Item):
return item
긴 설명이 필요한 경우 독스트링을 사용하여 경로 작동에 대한 설명을 작성할 수 있습니다. 독스트링은 마크다운 형식으로 작성할 수 있으며, 대화형 API 문서에서 마크다운으로 렌더링됩니다.
@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):
"""
Create an item with all the necessary information:
- **name**: The name of the item.
- **description**: A detailed description of the item.
- **price**: The price of the item.
- **tax**: The tax applied to the item, if any.
- **tags**: A set of unique tags associated with the item.
"""
return item
response_description 매개변수를 사용하여 응답에 대한 설명을 제공할 수 있습니다. 이는 대화형 문서에서 응답의 의미를 명확히 전달하는 데 도움이 됩니다.
@app.post(
"/items/",
response_model=Item,
summary="Create an item",
response_description="The created item will be returned",
)
async def create_item(item: Item):
return item
경로 작동이 더 이상 사용되지 않지만 여전히 제거하지는 않을 경우, deprecated 매개변수를 사용하여 이를 명시할 수 있습니다.
@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
return [{"item_id": "Foo"}]
경로 작동 데코레이터에서 제공하는 다양한 매개변수를 통해 경로 작동에 대한 메타데이터를 추가하거나, 응답 코드와 설명 등을 세밀하게 설정할 수 있습니다. 이를 통해 더 나은 API 문서와 사용성을 제공할 수 있으며, API의 가독성을 높일 수 있습니다.
때로는 Pydantic 모델과 같은 복잡한 데이터 유형을 JSON과 호환되는 형식으로 변환해야 하는 상황이 발생합니다. 이러한 변환이 필요한 이유 중 하나는 데이터베이스에 저장하거나, JSON 형식만을 수용하는 API와 통신할 때입니다.
FastAPI는 이러한 변환을 쉽게 처리할 수 있도록 jsonable_encoder 함수를 제공합니다. 이 함수는 Pydantic 모델, datetime 객체, 그리고 기타 JSON과 호환되지 않는 데이터 유형을 적절한 형식으로 변환합니다.
예를 들어, datetime 객체는 기본적으로 JSON과 호환되지 않기 때문에, 이를 JSON 형식으로 변환해야 할 필요가 있습니다. 마찬가지로 Pydantic 모델도 직접적으로 JSON으로 변환되지 않으므로, jsonable_encoder를 통해 JSON과 호환 가능한 형식으로 변환해야 합니다.
from datetime import datetime
from typing import Union
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
# 가상의 데이터베이스
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):
# Pydantic 모델을 JSON 호환 가능한 형식으로 변환
json_compatible_item_data = jsonable_encoder(item)
# 변환된 데이터를 fake_db에 저장
fake_db[id] = json_compatible_item_data
위 코드에서 jsonable_encoder는 Pydantic 모델을 dict로 변환하고, datetime 객체를 ISO 형식의 문자열로 변환합니다.
JSON은 웹 서비스와 데이터베이스에서 널리 사용되는 형식입니다. 그러나 Python의 datetime과 같은 객체는 JSON으로 직접 변환되지 않으며, JSON과 호환되지 않는 데이터 유형이 될 수 있습니다.
이를 해결하기 위해 jsonable_encoder는 이러한 객체들을 자동으로 변환합니다.
이 변환 과정을 통해 json.dumps() 함수로 직렬화 가능한 형태로 만들 수 있습니다.
FastAPI는 내부적으로 이 jsonable_encoder를 사용하여 데이터를 처리합니다. 하지만 사용자는 이 함수를 직접 호출하여 다양한 용도로 사용할 수 있습니다.
예를 들어, JSON과 호환되지 않는 데이터를 데이터베이스에 저장할 때나, API에 전달할 때 이를 사용할 수 있습니다.
아래 예시는 jsonable_encoder를 사용하여 Pydantic 모델을 JSON 호환 형식으로 변환한 후 가상 데이터베이스에 저장하는 방식입니다.
from datetime import datetime
from typing import Union
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
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):
# Pydantic 모델을 JSON 호환 가능한 데이터로 변환
json_compatible_item_data = jsonable_encoder(item)
# fake_db에 저장
fake_db[id] = json_compatible_item_data
이 예시에서 item 객체는 JSON과 호환 가능한 형태로 변환되어 데이터베이스에 저장됩니다. 이는 jsonable_encoder가 Pydantic 모델과 datetime 객체를 적절한 형식으로 변환하는 역할을 합니다.
jsonable_encoder는 JSON과 호환되지 않는 Python 객체를 변환하는 데 유용합니다.jsonable_encoder를 사용하여 JSON과 호환되지 않는 데이터를 손쉽게 변환하고, 이를 다양한 방식으로 활용할 수 있습니다.
HTTP PUT 메소드는 기존 데이터를 대체하는데 사용됩니다. 이 경우 jsonable_encoder를 사용하여 데이터를 JSON과 호환 가능한 형식으로 변환합니다. 예를 들어, datetime 객체는 문자열로 변환됩니다.
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str | None = None
description: str | None = None
price: float | None = None
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.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
PUT을 사용하면, 모든 데이터를 대체합니다. 즉, 일부 값만 제공하면 나머지는 기본값으로 덮어씁니다.
예를 들어, 기존 bar 아이템을 다음과 같이 업데이트하려고 한다면:
{
"name": "Barz",
"price": 3,
"description": null
}
이 경우 tax 속성을 포함하지 않았기 때문에 tax 값은 기본값인 10.5로 덮어쓰게 됩니다.
HTTP PATCH 메소드는 기존 데이터를 부분적으로 업데이트하는데 사용됩니다. PATCH를 사용하면 업데이트할 데이터만 전송하고 나머지는 그대로 남길 수 있습니다.
이때 Pydantic의 exclude_unset 매개변수를 사용하여 기본값이 설정되지 않은 필드만 업데이트할 수 있습니다.
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str | None = None
description: str | None = None
price: float | None = None
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.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)
update_data = item.dict(exclude_unset=True)
updated_item = stored_item_model.copy(update=update_data)
items[item_id] = jsonable_encoder(updated_item)
return updated_item
이 방법을 사용하면, 기본값을 덮어쓰지 않고, 사용자가 실제로 수정한 부분만 업데이트됩니다.
Pydantic에서는 model_copy() 메소드를 사용하여 기존 모델을 복사하고, 특정 필드를 업데이트할 수 있습니다. 이를 통해 부분 업데이트가 가능합니다.
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str | None = None
description: str | None = None
price: float | None = None
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.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)
update_data = item.dict(exclude_unset=True)
updated_item = stored_item_model.model_copy(update=update_data)
items[item_id] = jsonable_encoder(updated_item)
return updated_item
부분 업데이트를 구현하는 방법을 정리하면 다음과 같습니다:
1. PATCH 메소드 사용 (선택사항).
2. 저장된 데이터를 가져오기.
3. 해당 데이터를 Pydantic 모델에 넣기.
4. exclude_unset을 사용하여 입력된 데이터에서 기본값을 제외한 값만 선택.
5. 기존 모델을 복사하여 업데이트된 데이터로 속성 변경.
6. 업데이트된 데이터를 JSON과 호환 가능한 형식으로 변환하여 데이터베이스에 저장.
7. 업데이트된 모델 반환.
이 방법은 PUT 메소드에서도 사용할 수 있습니다. 그러나 PATCH는 부분 업데이트용으로 설계되었으므로, 부분적인 변경을 할 때 PATCH를 사용하는 것이 더 적합합니다.
업데이트를 받을 때에도 입력된 데이터는 여전히 유효성 검사를 거칩니다. 만약 모든 속성이 선택적이어야 한다면, 기본값을 None으로 설정하여 모든 필드를 선택적으로 만들 수 있습니다.
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class UpdateItemModel(BaseModel):
name: str | None = None
price: float | None = None
@app.patch("/items/{item_id}")
async def update_item(item_id: str, item: UpdateItemModel):
# 유효성 검사를 통과한 후 부분 업데이트
pass
이를 통해 생성과 업데이트에서 각기 다른 모델을 사용할 수 있습니다.
이와 같은 방법으로 FastAPI는 데이터 업데이트를 효율적으로 처리할 수 있습니다.
미들웨어는 FastAPI 애플리케이션에서 모든 요청과 모든 응답에 대해 경로 작동 전에 실행되고, 응답을 반환하기 전에 동작하는 함수입니다. 즉, 경로 함수 실행 전에 요청을 가로채어 무언가를 할 수 있고, 응답을 보내기 전에 그 응답을 조작할 수 있습니다.
미들웨어는 아래와 같은 순서로 동작합니다:
미들웨어는 @app.middleware("http") 데코레이터를 사용하여 정의할 수 있습니다. 미들웨어 함수는 두 가지 인자를 받습니다:
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request) # 요청을 경로 작업으로 전달
process_time = time.perf_counter() - start_time # 처리 시간 계산
response.headers["X-Process-Time"] = str(process_time) # 응답 헤더에 처리 시간 추가
return response # 응답 반환
이 예제에서는 요청 처리 시간을 측정하여 응답 헤더에 X-Process-Time이라는 사용자 정의 헤더로 추가합니다.
사용자 정의 헤더는 주로 'X-'로 시작합니다. 위의 예시처럼, X-Process-Time과 같은 사용자 정의 헤더를 추가할 수 있습니다. 하지만 이 헤더가 클라이언트(특히 브라우저)에서 접근 가능하게 하려면 CORS 설정에서 expose_headers 매개변수를 통해 노출시켜야 합니다.
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.perf_counter() # 요청 전 작업: 처리 시간 측정 시작
response = await call_next(request) # 요청 전달 및 응답 받음
process_time = time.perf_counter() - start_time # 처리 시간 계산
response.headers["X-Process-Time"] = str(process_time) # 응답 전 작업: 처리 시간 추가
return response # 응답 반환
미들웨어는 여러 개가 있을 수 있으며, FastAPI에서 제공하는 다른 미들웨어와도 조합하여 사용할 수 있습니다. 예를 들어, CORS 미들웨어와 함께 사용할 수 있습니다.
미들웨어에 대한 더 자세한 내용은 숙련된 사용자 안내서에서 확인할 수 있으며, 미들웨어와 함께 CORS를 처리하는 방법도 다룰 예정입니다.
이렇게 FastAPI에서 미들웨어를 활용하여 애플리케이션의 전역적인 기능을 쉽게 추가할 수 있습니다.
CORS (Cross-Origin Resource Sharing)는 브라우저에서 동작하는 프론트엔드 애플리케이션이 다른 출처의 백엔드와 HTTP 요청을 통해 데이터를 주고받을 수 있도록 허용하는 보안 메커니즘입니다.
출처는 다음과 같은 세 가지 요소로 구성됩니다:
1. 프로토콜: http, https 등
2. 도메인: myapp.com, localhost 등
3. 포트: 80, 443, 8080 등
예를 들어, 다음은 모두 서로 다른 출처입니다:
http://localhosthttps://localhosthttp://localhost:8080브라우저는 기본적으로 다른 출처에 대한 요청을 차단하기 때문에, CORS 설정이 필요합니다.
모든 출처에서 접근을 허용하려면, 와일드카드인 "*"을 사용할 수 있습니다. 그러나 쿠키나 인증 헤더(Authorization Header)를 사용하는 경우에는 와일드카드를 사용할 수 없으며, 반드시 특정 출처를 명시해야 합니다.
FastAPI에서 CORS 설정은 CORSMiddleware를 사용하여 쉽게 처리할 수 있습니다. 아래는 CORS 미들웨어를 설정하는 방법입니다:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# 허용된 출처 리스트
origins = [
"http://localhost.tiangolo.com",
"https://localhost.tiangolo.com",
"http://localhost",
"http://localhost:8080",
]
# CORS 미들웨어 추가
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # 허용된 출처들
allow_credentials=True, # 자격증명 허용 여부 (쿠키 등)
allow_methods=["*"], # 허용된 HTTP 메소드
allow_headers=["*"], # 허용된 HTTP 헤더
)
@app.get("/")
async def main():
return {"message": "Hello World"}
이 코드는 다양한 출처에서 GET, POST 등의 요청이 가능하게 하고, 자격증명을 허용하며 모든 HTTP 헤더와 메소드를 허용합니다.
['*']을 사용하여 모든 출처를 허용할 수 있습니다.['GET']이며, ['*']로 모든 메소드를 허용할 수 있습니다.[]이며, ['*']로 모든 헤더를 허용할 수 있습니다.False입니다.CORS 사전 요청 (OPTIONS):
단순 요청: