fastapi의 FastAPI() 메소드 사용법 정리
1. 애플리케이션 생성
1) FastAPI 인스턴스 생성
app = FastAPI(
title="My API", # API 제목
description="API 설명", # API 설명
version="1.0.0", # API 버전
docs_url="/docs", # Swagger UI 경로
redoc_url="/redoc", # ReDoc 경로
openapi_url="/openapi.json", # OpenAPI 스키마 경로
debug=False # 디버그 모드 설정
)
1. 매개변수 설명:
- title: API의 이름을 지정합니다. OpenAPI 문서에 표시된다.
- description: API에 대한 자세한 설명을 제공한다.
- version: API의 버전을 지정한다.
- docs_url: Swagger UI의 접근 경로를 설정합니다. None으로 설정하면 비활성화된다.
- redoc_url: ReDoc 문서의 접근 경로를 설정합니다. None으로 설정하면 비활성화된다.
- openapi_url: OpenAPI 스키마 JSON 파일의 경로를 설정한다.
- debug: 디버그 모드를 활성화/비활성화한다.
| 파라미터 | 기본값 | 설명 |
| title | "FastAPI" | OpenAPI 문서 제목 (예: title="My API") |
| description | "" | API 설명 (OpenAPI 문서에 표시) |
| version | "0.1.0" | API 버전 (예: version="1.0.0") |
| docs_url | "/docs" | Swagger UI 문서 URL (비활성화: docs_url=None) |
| redoc_url | "/redoc" | ReDoc 문서 URL (비활성화: redoc_url=None) |
| openapi_url | "/openapi.json" | OpenAPI 스키마 URL (비활성화: openapi_url=None) |
| debug | FALSE | 디버그 모드 활성화 (자세한 에러 메시지 표시) |
| dependencies | None | 전역 종속성 (예: dependencies=[Depend(common_dependency)]) |
2. 라우팅 메소드
1) HTTP 메소드 데코레이터
- 라우트를 간편하게 등록할 수 있도록 아래와 같은 데코레이터들을 제공합니다. 이들 데코레이터는 내부적으로 add_api_route()를 호출한다.
1. @app.get(path, …) : GET 요청 처리
2. @app.post(path, …) : POST 요청 처리
3. @app.put(path, …) : PUT 요청 처리
4. @app.delete(path, …) : DELETE 요청 처리
5. @app.patch(path, …) : PATCH 요청 처리
6. @app.options(path, …) : OPTIONS 요청 처리
7. @app.head(path, …) : HEAD 요청 처리
8. @app.trace(path, …) : TRACE 요청 처리
@app.get("/items/{item_id}")
@app.post("/items/")
@app.put("/items/{item_id}")
@app.delete("/items/{item_id}")
@app.patch("/items/{item_id}")
@app.options("/items/{item_id}")
@app.head("/items/{item_id}")
@app.trace("/items/{item_id}")
1. 매개변수 설명:
- path: 엔드포인트 경로
- response_model: 응답 데이터의 Pydantic 모델
- status_code: HTTP 상태 코드
- tags: API 문서화를 위한 태그
- summary: API 엔드포인트 요약
- description: 자세한 설명
- response_description: 응답에 대한 설명
- deprecated: 해당 엔드포인트의 사용 중단 여부
| 파라미터 | 타입 | 설명 |
| path | str | URL 경로 (기본값: 데코레이터가 적용된 함수의 경로) |
| status_code | int | 응답 상태 코드 (기본값: GET=200, POST=201, DELETE=204 등) |
| tags | List[str] | OpenAPI 문서에서 그룹화할 태그 (예: tags=["users"]) |
| summary | str | API 요약 설명 |
| description | str | API 상세 설명 |
| response_model | Type[BaseModel] | 응답 모델 (Pydantic 모델) |
| deprecated | bool | API 사용 중단 표시 (OpenAPI 문서에서 강조) |
| dependencies | List[Depends] | 라우트별 종속성 (예: dependencies=[Depend(auth)]) |
| response_class | Response | 기본 응답 클래스 (예: JSONResponse, HTMLResponse) |
| responses | Dict | OpenAPI 문서에 추가할 응답 예시 (예: {404: {"description": ...}}) |
| include_in_schema | bool | OpenAPI 스키마 포함 여부 (기본값: True) |
from fastapi import FastAPI, status
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
app = FastAPI(title="My API", description="API 예제", version="1.0.0")
@app.get("/items/{item_id}", response_model=Item, status_code=status.HTTP_200_OK, tags=["items"],
summary="아이템 조회", description="주어진 item_id에 해당하는 아이템 정보를 반환합니다.")
async def read_item(item_id: int):
# 실제 데이터 조회 로직
return {"name": "Sample Item", "price": 42.0}@app.get(
"/items/{item_id}",
response_model=Item,
status_code=200,
tags=["items"],
summary="아이템 조회",
description="ID를 기반으로 특정 아이템을 조회합니다.",
response_description="요청한 아이템 정보"
)
async def read_item(item_id: int):
return {"item_id": item_id}
2. 경로 파라미터 & 쿼리 파라미터
- 라우트 함수에서 사용되는 파라미터 (예: @app.get("/items/{item_id}")).
| 파라미터 유형 | 예시 | 설명 |
| 경로 파라미터 | item_id: int | URL 경로에서 추출 ({item_id}) |
| 쿼리 파라미터 | skip: int = 0 | URL 쿼리 문자열에서 추출 |
| 요청 바디 | item: Item (Pydantic 모델) | POST/PUT 요청의 본문 데이터 |
| Form 데이터 | username: str = Form(...) | HTML Form 데이터 수신 |
| 파일 업로드 | file: UploadFile = File(...) | 파일 업로드 처리 |
2) add_api_route() 메소드
- add_api_route()는 모든 HTTP 메소드 데코레이터의 기반이 되는 메소드이다. 직접 호출하여 라우트를 등록할 수도 있다.
- 파라미터
- path (str): 라우트 경로
- endpoint (Callable): 요청 시 호출되는 함수(핸들러)
- methods (Optional[List[str]]): 허용할 HTTP 메소드 목록 (예: ["GET"])
- response_model (Optional[Type]): 응답으로 반환할 데이터의 Pydantic 모델
- 자동으로 데이터 검증, 직렬화, OpenAPI 문서 생성에 사용된다.
- status_code (Optional[int]): 기본 응답 상태 코드
- tags (Optional[List[str]]): API 문서에서 그룹핑할 태그
- summary (Optional[str]): 간략한 요약
- description (Optional[str]): 상세 설명 (docstring 형식도 지원)
- response_description (str): 응답에 대한 설명 (기본값: "Successful Response")
- responses (Optional[Dict[Union[int, str], Dict[str, Any]]]): 추가 응답 설명
- deprecated (bool): 해당 엔드포인트가 더 이상 사용되지 않음을 명시할 때
- name (Optional[str]): 내부적으로 사용할 이름 (OpenAPI 스키마에 반영)
- include_in_schema (bool): OpenAPI 문서에 포함할지 여부 (기본값: True)
- response_class (Optional[Type[Response]]): 응답 클래스를 지정 (예: JSONResponse, HTMLResponse)
- dependencies (Optional[Sequence[Depends]]): 엔드포인트 전반에 적용할 의존성 목록
- callbacks (Optional[List[APIRoute]]): 콜백 경로 등
- 예시:
def my_endpoint():
return {"message": "Hello"}
app.add_api_route(
path="/hello",
endpoint=my_endpoint,
methods=["GET"],
response_model=dict,
status_code=200,
tags=["example"],
summary="Hello Endpoint",
description="간단한 인사말을 반환합니다.",
)
3) include_router() 메소드
- 복잡한 애플리케이션에서는 여러 개의 라우터(APIRouter)를 분리하여 관리할 수 있다.
- include_router()를 사용하면 외부에서 정의된 라우터를 현재 애플리케이션에 포함시킬 수 있다.
- 주요 파라미터
- router (APIRouter): 포함할 라우터 객체
- prefix (Optional[str]): 라우터 내 모든 경로 앞에 붙일 공통 경로
- tags (Optional[List[str]]): 해당 라우터에 적용할 태그 목록
- dependencies (Optional[Sequence[Depends]]): 라우터 전역에 적용할 의존성 목록
- responses (Optional[Dict]): 라우터에 적용할 응답 설정
| 파라미터 | 설명 |
| router | 포함할 APIRouter 객체 |
| prefix | 라우터 경로에 추가할 접두사 (예: prefix="/api") |
| tags | 라우터의 모든 경로에 적용될 태그 (예: tags=["auth"]) |
| dependencies | 라우터 전체에 적용될 종속성 |
| responses | 라우터의 모든 경로에 적용될 공통 응답 |
from fastapi import APIRouter
router = APIRouter()
@router.get("/users")
async def get_users():
return [{"username": "alice"}, {"username": "bob"}]
app.include_router(router, prefix="/api", tags=["users"])
4) WebSocket 라우트 등록
1. @app.websocket() 데코레이터 사용하기
- 주요 특징
- 직관적 정의: 일반적인 HTTP 엔드포인트와 유사하게 데코레이터를 사용해 WebSocket 라우트를 정의할 수 있다.
- 비동기 처리: WebSocket 핸들러는 반드시 비동기 함수(async def)로 작성해야 하며, 연결 수락, 메시지 송수신 등 모든 작업이 비동기적으로 이루어진다.
- 동작 과정
- 연결 수락:
- await websocket.accept()를 호출하여 클라이언트의 연결 요청을 승인한다.
- 메시지 송수신:
- 무한 루프 내에서 receive_text() (또는 receive_bytes())를 통해 데이터를 기다리고, 수신한 데이터를 send_text()로 클라이언트에 응답한다.
- 예외 처리:
- 클라이언트가 연결을 끊으면 WebSocketDisconnect 예외가 발생하므로 이를 캐치하여 종료 처리한다.
- 기본 코드 예제
from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
# 클라이언트의 연결 요청을 수락
await websocket.accept()
try:
while True:
# 클라이언트로부터 텍스트 메시지 수신 (또는 receive_bytes()로 이진 데이터)
data = await websocket.receive_text()
# 받은 메시지를 가공하여 다시 전송
await websocket.send_text(f"Echo: {data}")
except WebSocketDisconnect:
# 연결 종료 시 처리할 로직
print("클라이언트 연결 종료")
2. app.add_websocket_route() 메소드 사용하기
- 주요 특징
- 프로그램적 등록: 데코레이터 대신 함수나 클래스를 사용해 WebSocket 라우트를 동적으로 등록할 수 있다.
- 유연한 구성: 여러 라우트를 프로그램적으로 추가하거나, 라우트 구성이 동적으로 변경되어야 할 경우 유용하다.
- 동작 과정
- 핸들러 함수 정의:
- 데코레이터를 사용하지 않고 별도의 함수(websocket_endpoint)를 정의한다.
- 라우트 등록:
- app.add_websocket_route("/ws", websocket_endpoint)를 호출하여 지정한 경로에 해당 핸들러를 등록한다.
- 기능은 동일:
- 등록된 핸들러는 데코레이터로 등록한 경우와 동일하게 클라이언트의 연결 수락, 메시지 처리, 예외 처리 등을 수행한다.
- 기본 코드 예제
from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocketDisconnect
app = FastAPI()
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data}")
except WebSocketDisconnect:
print("클라이언트 연결 종료")
# '/ws' 경로에 대해 websocket_endpoint 함수를 등록
app.add_websocket_route("/ws", websocket_endpoint)
3. 두 방법의 비교 및 선택 가이드
- @app.websocket() 데코레이터
- 장점: 코드가 간결하고 직관적이며, 정적 라우트 정의에 적합하다.
- 사용 사례: 대부분의 경우 고정된 WebSocket 엔드포인트를 구현할 때 주로 사용한다.
- app.add_websocket_route() 메소드
- 장점: 라우트를 동적으로 추가하거나, 모듈화된 방식으로 관리할 때 유용하다.
- 사용 사례: 라우트 구성을 프로그래밍적으로 제어해야 하거나, 조건에 따라 다수의 WebSocket 엔드포인트를 등록할 때 적합하다.
4. 결론
- FastAPI(또는 Starlette)에서는 WebSocket 연결을 지원하는 두 가지 방법, 데코레이터 방식과 메소드 방식을 제공하며, 두 방법 모두 동일한 기본 원리를 따른다.
- @app.websocket() 데코레이터는 단순하고 직관적인 문법으로 고정된 엔드포인트를 정의할 때 유리하다.
- app.add_websocket_route() 메소드는 보다 동적인 라우트 등록이 필요할 때 효과적이다.
- 이 두 방법 중 프로젝트의 요구사항과 코드 구성 방식에 맞춰 적절한 방법을 선택하여 사용하면 됩니다.
3. 미들웨어 관련 메소드
1) 일반적인 커스텀 미들웨어 추가
- add_middleware의 첫번째 파라미터에 정의되는 메소드의 __init__ 메서드에서 정의된 매개변수이다.
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# 미들웨어 함수 정의
async def custom_middleware(request: Request, call_next):
# 요청 전처리 로직
print("Custom middleware processing before request.")
# 다음 미들웨어나 핸들러 호출
response = await call_next(request)
# 응답 후처리 로직
print("Custom middleware processing after response.")
return response
# 미들웨어를 FastAPI 앱에 추가
app.add_middleware(custom_middleware)
2) CORS 미들웨어를 설정
# CORS 미들웨어 설정
origins = [
"http://localhost",
"http://localhost:8000",
]
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com"], # 허용할 오리진 목록
allow_credentials=True, # 자격 증명 허용 여부
allow_methods=["GET", "POST"], # 허용할 HTTP 메서드
allow_headers=["Authorization", "Content-Type"], # 허용할 HTTP 헤더
)
4. 예외 처리 메소드
1) 예외 핸들러 등록
1. 커스텀 예외 클래스 정의
- 먼저, 애플리케이션에서 발생할 수 있는 특정 상황을 처리하기 위해 사용자 정의 예외 클래스를 만든다.
- CustomException 클래스는 예외 발생 시 이름을 전달받아 초기화 한다.
class CustomException(Exception):
def __init__(self, name: str):
self.name = name
2. 예외 핸들러 함수 작성
- 다음으로, 위에서 정의한 CustomException을 처리할 핸들러 함수를 작성한다.
- 이 핸들러는 CustomException이 발생할 때 호출되며, JSON 형식의 응답을 반환한다.
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(
status_code=418, # I'm a teapot 상태 코드
content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
)
3. 예외 핸들러 등록
- FastAPI 애플리케이션 인스턴스에 예외 핸들러를 등록한다.
from fastapi import FastAPI
app = FastAPI()
app.add_exception_handler(CustomException, custom_exception_handler)
4. 엔드포인트에서 예외 발생
- 이제 엔드포인트에서 CustomException을 발생시켜 예외 핸들러가 작동하는지 확인한다.
@app.get("/items/{name}")
async def read_item(name: str):
if name == "rainbow":
raise CustomException(name=name)
return {"name": name}
5. 전체 코드
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class CustomException(Exception):
def __init__(self, name: str):
self.name = name
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
)
app = FastAPI()
app.add_exception_handler(CustomException, custom_exception_handler)
@app.get("/items/{name}")
async def read_item(name: str):
if name == "rainbow":
raise CustomException(name=name)
return {"name": name}
6. add_exception_handler 없이 exception_handler를 이용해 간편하게 등록하는 방법
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class CustomException(Exception):
def __init__(self, name: str):
self.name = name
app = FastAPI()
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(
status_code=400,
content={"message": f"An error occurred: {exc.name}"},
)
@app.get("/items/{name}")
async def read_item(name: str):
if name == "error":
raise CustomException(name=name)
return {"name": name}
5. 이벤트 핸들러 메소드
1) @app.on_event 데코레이터를 사용한 이벤트 핸들러
- 이전에는 @app.on_event 데코레이터를 사용하여 이벤트 핸들러를 정의했다.
- 이 방법은 여전히 사용 가능하지만, FastAPI의 최신 버전에서는 lifespan 매개변수를 사용하는 것이 권장된다.
- startup_event 함수는 애플리케이션이 시작될 때, shutdown_event 함수는 애플리케이션이 종료될 때 호출된다.
from fastapi import FastAPI
app = FastAPI()
@app.on_event("startup")
async def startup_event():
# 애플리케이션 시작 시 실행할 코드
print("애플리케이션이 시작되었습니다.")
@app.on_event("shutdown")
async def shutdown_event():
# 애플리케이션 종료 시 실행할 코드
print("애플리케이션이 종료되었습니다.")from fastapi import FastAPI
# FastAPI 애플리케이션 인스턴스 생성
app = FastAPI()
# 데이터베이스 커넥션 예시 변수 (실제 코드에서는 데이터베이스 라이브러리 사용)
db_connection = None
# 애플리케이션 시작 시 실행될 이벤트 핸들러
@app.on_event("startup")
async def startup_event():
global db_connection
# 데이터베이스 연결 초기화 (예시 코드)
db_connection = "DB 연결 생성"
# 실제 코드에서는 await some_async_db_connect() 등 사용
print("애플리케이션 시작: 데이터베이스 연결이 초기화되었습니다.")
# 애플리케이션 종료 시 실행될 이벤트 핸들러
@app.on_event("shutdown")
async def shutdown_event():
global db_connection
# 데이터베이스 연결 종료 (예시 코드)
db_connection = None
# 실제 코드에서는 await some_async_db_disconnect() 등 사용
print("애플리케이션 종료: 데이터베이스 연결이 종료되었습니다.")
# 간단한 엔드포인트 예제
@app.get("/")
async def read_root():
return {"message": "Hello, World!"}
2) lifespan 매개변수를 사용한 이벤트 핸들러
- FastAPI 0.65.0 버전부터는 lifespan 매개변수를 사용하는 것이 권장됩니다. 이를 통해 애플리케이션의 시작과 종료 시점에 실행할 코드를 더 명확하게 정의할 수 있다.
- lifespan 함수는 @asynccontextmanager 데코레이터를 사용하여 정의되며, yield 이전의 코드는 애플리케이션 시작 시, yield 이후의 코드는 애플리케이션 종료 시 실행된다.
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# 애플리케이션 시작 시 실행할 코드
print("애플리케이션이 시작되었습니다.")
yield
# 애플리케이션 종료 시 실행할 코드
print("애플리케이션이 종료되었습니다.")
app = FastAPI(lifespan=lifespan)from fastapi import FastAPI
from contextlib import asynccontextmanager
# 예시: 리소스(예: 데이터베이스 커넥션)를 관리할 변수
db_connection = None
# lifespan 함수 정의 (애플리케이션의 시작과 종료 시점에 호출됨)
@asynccontextmanager
async def lifespan(app: FastAPI):
global db_connection
# 애플리케이션 시작 시 실행되는 코드
db_connection = "DB 연결 생성 (lifespan)"
# 실제 코드에서는 await some_async_db_connect() 사용
print("Lifespan 시작: 데이터베이스 연결이 초기화되었습니다.")
# yield 이전의 코드는 startup 시점에, yield 이후의 코드는 shutdown 시점에 실행됩니다.
try:
yield
finally:
# 애플리케이션 종료 시 실행되는 코드
db_connection = None
# 실제 코드에서는 await some_async_db_disconnect() 사용
print("Lifespan 종료: 데이터베이스 연결이 종료되었습니다.")
# FastAPI 애플리케이션 인스턴스 생성 시 lifespan 함수를 등록
app = FastAPI(lifespan=lifespan)
# 간단한 엔드포인트 예제
@app.get("/")
async def read_root():
return {"message": "Hello, World!"}
6. OpenAPI/Swagger 관련 메소드
1) FastAPI 생성 시 기본 설정
- FastAPI 애플리케이션 인스턴스를 생성할 때 여러 매개변수를 통해 OpenAPI 스키마와 문서화 엔드포인트를 제어할 수 있다.
1. openapi_url
- OpenAPI 스키마 JSON 파일의 URL을 지정한다.
- 예) openapi_url="/openapi.json"
- 값을 None으로 지정하면 OpenAPI 스키마 생성을 비활성화할 수 있다.
2. docs_url
- Swagger UI 문서 페이지의 URL을 지정한다.
- 예) docs_url="/docs"
- 값을 None으로 지정하면 Swagger UI를 비활성화할 수 있다.
3. redoc_url
- ReDoc 문서 페이지의 URL을 지정한다.
- 예) redoc_url="/redoc"
- 값을 None으로 지정하면 ReDoc을 비활성화할 수 있다.
4. swagger_ui_parameters
- Swagger UI의 동작 및 테마와 같은 추가 옵션을 설정할 수 있다.
- 예) swagger_ui_parameters={"syntaxHighlight.theme": "monokai"}
from fastapi import FastAPI
app = FastAPI(
title="My API",
description="This is a sample API.",
version="1.0.0",
openapi_url="/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
swagger_ui_parameters={"syntaxHighlight.theme": "monokai"},
)
2. OpenAPI 스키마 생성 및 커스터마이징
1) 기본 OpenAPI 스키마 자동 생성
- FastAPI는 애플리케이션의 모든 경로와 모델(Pydantic)을 분석하여 OpenAPI 스키마를 자동 생성한다. 이 스키마는 /openapi.json 엔드포인트에서 확인할 수 있다.
2) get_openapi() 함수
- fastapi.openapi.utils.get_openapi 함수를 사용하면 현재 앱의 속성(예: title, version, description, routes)을 바탕으로 OpenAPI 스키마를 생성할 수 있다. 이 함수를 오버라이드하여 사용자 정의 스키마를 만들 수 있다.
- 이렇게 재정의하면, /openapi.json에 반영된 스키마에 사용자 정의 필드나 변경 사항을 추가할 수 있다.
from fastapi.openapi.utils import get_openapi
def custom_openapi():
if app.openapi_schema: # 이미 생성된 경우 캐시된 스키마 반환
return app.openapi_schema
openapi_schema = get_openapi(
title=app.title,
version=app.version,
description=app.description,
routes=app.routes,
)
# 필요한 커스터마이징을 추가할 수 있음
openapi_schema["info"]["x-custom"] = "Custom value"
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
3. 추가 커스터마이징 옵션
1) openapi_tags
- 라우터 데코레이터의 tags 인자를 통해 엔드포인트를 그룹화하면, 자동 생성된 문서에서 태그별 섹션으로 정리되어 표시된다.
@app.get("/items/", tags=["items"])
async def read_items():
return [{"item_id": "Foo"}]
2) 요약 및 설명
- 각 엔드포인트에 summary와 description 매개변수를 지정하면, 문서에 해당 설명이 반영되어 사용자가 엔드포인트의 역할을 쉽게 이해할 수 있다.
@app.post("/items/", summary="아이템 생성", description="새로운 아이템을 생성합니다.")
async def create_item(item: Item):
return item
7. CORS 설정 메소드
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:8080", "https://example.com"],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["*"],
max_age=3600
)
- 매개변수:
- allow_origins: 허용할 출처 목록
- allow_credentials: 인증 정보 허용 여부
- allow_methods: 허용할 HTTP 메소드 목록
- allow_headers: 허용할 HTTP 헤더 목록
- max_age: 프리플라이트 요청 캐시 시간(초)
8. 서브 애플리케이션 및 정적 파일 마운트
1) mount() 메소드
- mount()를 사용하면 별도의 ASGI 애플리케이션(예: Starlette의 StaticFiles)을 특정 경로에 "마운트"할 수 있다.
- 예시 (정적 파일 제공):
from starlette.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")
9. 백그라운드 태스크 실행
- FastAPI는 Starlette의 BackgroundTasks 클래스를 활용하여 백그라운드 태스크를 지원한다. 사용 방법은 다음과 같다.
1) 백그라운드 태스크 인스턴스 주입:
- 경로 함수의 매개변수로 BackgroundTasks 타입을 선언하면 FastAPI가 자동으로 인스턴스를 생성하여 주입한다.
2) 태스크 등록:
- background_tasks.add_task() 메서드를 사용하여 실행하고자 하는 함수를 등록한다. 이때 인자나 키워드 인자도 함께 전달할 수 있다.
3) 즉시 응답 반환:
- 태스크가 등록되면 함수는 바로 응답을 반환한다. 등록된 태스크는 응답이 전송된 후에 순차적으로 실행된다.
4) 실행 시점:
- FastAPI는 HTTP 응답이 성공적으로 클라이언트에 전송된 후에, 등록된 모든 백그라운드 태스크를 실행한다.
- 만약 경로 함수 내에서 예외가 발생하여 응답을 반환하지 못하면, 등록된 백그라운드 태스크는 실행되지 않는다.
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
# 동기 함수 예시 (def로 정의)
def write_log(message: str):
with open("log.txt", "a") as file:
file.write(message + "\n")
@app.get("/")
async def read_root(background_tasks: BackgroundTasks):
# 백그라운드 태스크 등록
background_tasks.add_task(write_log, "로그 메시지")
return {"message": "작업이 백그라운드에서 실행됩니다."}
10. dependency_overrides를 이용한 의존성 주입
- FastAPI의 의존성 주입 시스템은 테스트나 특정 환경에서 실제 의존성 대신 다른 동작을 주입할 수 있도록 dependency_overrides 기능을 제공한다.
- 이를 통해 실제 리소스(예: 데이터베이스 연결, 외부 API 호출 등)를 사용하지 않고도, 모의 객체(mock)나 간소화된 로직을 사용하여 테스트를 진행하거나 환경에 맞게 동작을 변경할 수 있다.
1) 사용법
1. 오버라이딩 함수 정의:
- 원래 의존성이 제공하는 값을 대체할 함수를 작성한다.
2. 오버라이딩 등록:
- 앱 객체의 dependency_overrides에 원래 함수와 대체 함수를 등록한다.
app.dependency_overrides[원래_의존성함수] = 오버라이딩_함수
3. 테스트 또는 실행 시 활용:
- 이후 해당 의존성이 필요한 경로 함수는 오버라이딩된 함수의 결과를 사용한다.
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
app = FastAPI()
# 실제 DB 연결을 위한 의존성 함수
def get_db():
return "Real DB Connection"
@app.get("/items/")
def read_items(db=Depends(get_db)):
return {"db": db}
# 테스트를 위해 DB 연결 대신 사용할 오버라이딩 함수
def override_get_db():
return "Fake DB Connection"
# dependency_overrides에 오버라이딩 함수 등록
app.dependency_overrides[get_db] = override_get_db
# TestClient를 사용하여 테스트 실행
client = TestClient(app)
def test_read_items():
response = client.get("/items/")
assert response.json() == {"db": "Fake DB Connection"}
# 테스트 실행 시, read_items 엔드포인트는 get_db 대신 override_get_db의 결과를 사용하게 됩니다.
2) 주요 포인트
1. 테스트 용이성:
- 실제 데이터베이스나 외부 서비스에 의존하지 않고 테스트 할 수 있으므로, 테스트 속도와 안정성이 향상된다.
2. 유연성:
- 특정 환경(예: 개발, 프로덕션, 테스트)에 따라 동일한 경로 함수라도 다른 동작을 하도록 설정할 수 있다.
3. 전역 적용:
- app.dependency_overrides에 등록된 오버라이딩은 전역적으로 적용되므로, 모든 요청에서 해당 의존성이 오버라이딩된 함수로 대체된다.
- reference :
https://fastapi.tiangolo.com/ko/advanced/middlewares/
고급 미들웨어 - FastAPI
FastAPI framework, high performance, easy to learn, fast to code, ready for production
fastapi.tiangolo.com
[FastAPI] Lifespan을 이용한 생명주기 관리 (Event Handler)
서론 이번 글에서는 FastAPI의 Lifespan을 이용하여 이벤트를 관리하는 방법을 알아보고자 한다. 이 글에서 말하는 이벤트 핸들링은 FastAPI 서비스가 시작 또는 종료될 때 특정 함수를 호출하는 등의
limsw.tistory.com
https://fastapi.tiangolo.com/ko/advanced/events/
이벤트: startup과 shutdown - FastAPI
FastAPI framework, high performance, easy to learn, fast to code, ready for production
fastapi.tiangolo.com
https://fastapi.tiangolo.com/advanced/events/
Lifespan Events - FastAPI
FastAPI framework, high performance, easy to learn, fast to code, ready for production
fastapi.tiangolo.com
[FastAPI] 이벤트 기반 아키텍처 구축하기 (w. fastapi-events)
FastAPI 프레임워크로 이벤트 기반 아키텍처를 구축하는 방법에 대해 공유한다. # 이벤트 기반 아키텍처란? 새로운 댓글 작성 시, 게시물 작성자에게 알림을 보내야 하는 기능을 구현해야 한다고
mopil.tistory.com
'Study > fastapi' 카테고리의 다른 글
| vscode, cusor ai에서 fastapi debug 하는 방법 with launch.json (0) | 2025.02.19 |
|---|---|
| FastAPI에서 SQLAlchemy를 사용하여 테이블을 생성하는 방법 (0) | 2025.02.17 |
| [udemy] Try FastAPI Test Driven Development 2024 - 학습 후기 (0) | 2025.02.15 |
| [udemy] Try FastAPI Test Driven Development 2024 - 학습 정리 (0) | 2025.02.15 |
| tenacity를 이용한 retry 사용 방법 정리 (0) | 2025.02.04 |
댓글