- Python으로 RESTful API를 쉽고 빠르게 개발하는 모던 웹 프레임워크
- 비동기 지원
- 웹 애플리케이션 : CRUD API, 대시보드, 비즈니스 로직 처리
- 마이크로 서비스 : 비동기와 경량 설계로 효율적인 마이크로서비스 작성
- 머신러닝/AI 배포 : FastAPI를 통해 AI모델을 웹 API로 간단히 배포
비동기 처리(Asynchronous Programming)
데이터 검증 및 직렬화
경량설계
타입 힌트 기반 개발
uvicorn main:app --reload
from fastapi import FastAPI
# FastAPI 애플리케이션 인스턴스 생성
app = FastAPI()
# 루트 엔드포인트 정의 (GET 요청)
@app.get("/")
def hello():
# JSON 응답 반환
return {"msg" : "Hello, FastAPI"}
#### 쿼리 매개변수
> * URL에서 `?key=value` 형식으로 전달되는 값
> * 경로 매개변수와 달리 선택적으로 사용
> * 경로에 넣지 않고 함수 자체에 넣음
```python
@app.get("/items/")
def read_items(skip: int= 0,limit: int = 10):
return {"skip":skip, "limit": limit}
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"item_id" :item_id}
/items/?skip=5&limit=15
@app.get("/items/")
def read_item(skip: int=0, limit: int = 10):
return {"skip": skip, "limit": limit}
import asyncio
async def task1():
print("🚀 Task 1 시작")
await asyncio.sleep(3) # ✅ 3초 동안 멈춤 (다른 작업 실행 가능)
print("✅ Task 1 완료")
async def task2():
print("🔥 Task 2 시작")
await asyncio.sleep(2) # ✅ 2초 동안 멈춤
print("✅ Task 2 완료")
async def main():
await asyncio.gather(task1(), task2()) # ✅ 두 개의 비동기 함수를 동시에 실행
asyncio.run(main())
# ✔ "🚀 Task 1 시작" & "🔥 Task 2 시작" → 동시에 실행됨
# ✔ 2초 후 "✅ Task 2 완료" 출력 (task2()가 먼저 종료됨)
# ✔ 1초 후 "✅ Task 1 완료" 출력 (task1()가 이후에 종료됨)
구조,
pip install jinja2
: Jinja2는 /templates 폴더에서 템플릿을 자동으로 찾습니다./my_fastapi_project │── main.py # FastAPI 앱 실행 파일 │── /templates # HTML 템플릿 폴더 │ ├── index.html # 기본 HTML 파일 │ ├── about.html # 추가 HTML 파일 │── /static # CSS, JavaScript, 이미지 파일 │ ├── styles.css │── requirements.txt # 설치해야 할 패키지 리스트
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
app = FastAPI()
# 템플릿 폴더 설정
templates = Jinja2Templates(directory="templates")
# 정적 파일 (CSS, JS 등) 제공
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/")
async def home(request: Request):
return templates.TemplateResponse("index.html", {"request": request, "title": "FastAPI + Jinja2!"})
@app.get("/about")
async def about(request: Request):
return templates.TemplateResponse("about.html", {"request": request, "title": "About Page"})
templates/index.html
파일 추가
- Pydantic은 FastAPI에서 데이터 검증과 직렬화(Serialization)를 쉽게 처리할 수 있도록 도와주는 라이브러리
- Python의 데이터 유효성 검사 및 데이터 변환을 쉽게 처리
- FastAPI의 요청(Request) 및 응답(Response) 데이터를 검증할 때 필수적으로 사용됨
- 타입 힌트를 기반으로 데이터 검증을 자동으로 수행!
from pydantic import BaseModel
class Item(BaseModel):
name :str
price :float
if_offer : bool = False
@app.post("/items/")
def create_item(item: Item):
return {"item": item}
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str # 필수 필드 (기본값 없음)
price: float = Field(gt=0) # 필수 필드 (양수만 가능)
description: str = "No description" # 선택적 필드 (기본값 있음)
#gt >
#lt <
#ge >=
#le <=
class Reservation(BaseModel):
name : str = Field(..., max_length = 50, description="Reservation Name")
email : str
date : datetime
special_requests : str = ""
@field_validator("date")
@classmethod
def validate_date(cls, value):
date_obj = value
if date_obj < datetime.now():
raise ValueError("Reservation Must Be Future")
return value
class ExampleMode(BaseModel):
number: int # 이게에 대하여 좀더 구체적으로 잡고싶으면 field_validator 사용
@field_validator("number") # 뭘 구체적으로할건지
@classmethod # 이것도 무조건 적어야함
def validate_number(cls, value): # cls 도 적어야함 value가 자동으로 number로 맵핑된다
if value < 1 :
raise ValueError("Number must be at least 1")
if value > 100 :
raise ValueError("Number must not exceed 100")
return value
- 선택적 값, 기본값이 None일 수 있음
- Optional[str]을 사용하면 쿼리 매개변수가 없어도 허용됨
- 설명하지 않으면 기본값이 없는 파라미터는 필수(required)
- [GET] /items/?names=hello&names=world
- Optional은 해당 필드가 None 값을 가질 수 있다는 것을 명시하는 타입 힌트
from fastapi import Query
@app.get("/items/")
def read_items(names: List[str] = Query(default =[]))): # Query를 쓰지않으면 list 처리가 안됨
return {"names" : names}
@app.get("/items/")
def get_config(settings: str = Query('{"theme":"dark"}')): # JSON 문자열로 받음
return json.loads(settings)
List[T]
: 쿼리 매개변수 및 POST 요청에서 여러 개의 값을 받을 때[GET]
/items/?names=hello&names=worlddef read_items(names: List[str] =Query(default=[])):
Dict[str, T]
: JSON 데이터를 받을 때[GET]
/config/?settings={"theme": "light"}def get_config(settings: str = Query('{theme":"dark"}')): # Json문자열로 받음
구조
동작
- Swagger UI :
/docs
- ReDoc :
/redoc
- Swagger UI
- FastAPI에서 자동 생성되는 API 문서
- OpenAPI 명세 기반으로 작동
- 주요 기능
- 시각화: 모든 API 엔드포인트와 요청/응답을 쉽게 확인
- 테스트 가능: API 요청을 직접 실행하고 응답 확인
- ReDoc
- Swagger UI의 대안으로 제공되는 API 문서화 도구
- OpenAPI 명세란?
- RESTfual API의 정의를 표준화한 문서 형식
- API의 엔드포인트, 요청/응답 데이터, 상태 코드, 인증방식을 정의
- JSON 또는 YAML 형식으로 작성
- Swagger UI or Redoc 으로 명세
app = FastAPI(
title="My FastAPI Project",
description="This is a sample API to demonstrate Swagger UI and OpenAPI features.
version="1.0.0",
contact={
"name": "API Support",
"email": "support@example.com",
},
license_info={
"name": "MIT License"
"url": "https://opensource.org/licenses/MIT"',
},
}
# 태그
@app get("/users/", tags=["Users"])
def get_users ():
return [{"name": "John"}, {"name": "Jane"}]
@app.get ("/products/", tags=["Products"])
def get_products ():
return [{"name": "Laptop"}, {"'name": "Phone}]
# 설명
@app. get (
"/orders/{order_id}",
summary="Retrieve an order",
description="Retrieve the details of an order by its unique ID."
)
def get_order(order_id: int):
return {"order_id": order_id, "status": "Shipped"}
# 엔드포인트 변경
app = FastAPI(docs_url="/my-docs")
# API 버전별 문서 : 별도의 FastAPI 인스턴스를 생성
from fastapi import APIRouter # 라우터
v1 = APIRouter()
v2 = APIRouter()
app.include_router(v1, prefix="/v1", tags=["v1"])
app.include_router(v2, prefix="/v2", tags=["v2"])
요청과 응답 사이에서 실행되는 함수, 모든 요청이 들어오거나 응답이 나나기 전에 실행되는 중간 처리 단계
from fastapi import FastAPI, Request
app = FastAPI()
# 첫 번째 미들웨어
@app.middleware("http")
async def first_middleware(request: Request, call_next):
print("🔵 첫 번째 미들웨어: 요청 전")
response = await call_next(request) # 다음 미들웨어 or 엔드포인트 실행
print("🔵 첫 번째 미들웨어: 응답 후")
return response
# 두 번째 미들웨어
@app.middleware("http")
async def second_middleware(request: Request, call_next):
print("🟢 두 번째 미들웨어: 요청 전")
response = await call_next(request) # 최종적으로 엔드포인트 실행됨
print("🟢 두 번째 미들웨어: 응답 후")
return response
# 엔드포인트 (최종 실행)
@app.get("/")
async def home():
print("✅ 엔드포인트 실행됨")
return {"message": "Hello FastAPI"}
# 🔵 첫 번째 미들웨어: 요청 전
# 🟢 두 번째 미들웨어: 요청 전
# ✅ 엔드포인트 실행됨
미들웨어를 데코레이터(@app.middleware("http"))로 등록하면, 클라이언트 요청이 오면 먼저 미들웨어 내부의 call_next 이전 코드가 실행되고, call_next(request)가 호출되면 다음 미들웨어나 최종적으로 FastAPI의 엔드포인트(@app.get(), @app.post())로 요청이 전달됨.
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from datetime import datetime, timedelta
import jwt
# JWT 설정값
SECRET_KEY = "mysecretkey" # 안전한 환경에서는 환경변수로 설정
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 토큰 만료 시간
app = FastAPI()
# JWT 토큰 생성 함수
def create_jwt_token(data: dict):
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
data.update({"exp": expire}) # 만료 시간 추가
return jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM)
# 로그인 엔드포인트 (JWT 발급)
@app.post("/login")
async def login(username: str, password: str):
# 실제 서비스에서는 DB에서 유저 정보 확인해야 함
if username == "admin" and password == "password":
token = create_jwt_token({"sub": username}) # JWT 생성
return {"access_token": token, "token_type": "bearer"}
raise HTTPException(status_code=401, detail="Invalid credentials")
# JWT 인증 미들웨어
@app.middleware("http")
async def jwt_auth_middleware(request: Request, call_next):
if request.url.path in ["/login", "/docs", "/openapi.json"]: # 로그인, 문서 API는 예외
return await call_next(request)
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
return HTTPException(status_code=401, detail="Token missing or invalid")
token = auth_header.split(" ")[1] # "Bearer {token}" 에서 토큰만 추출
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
request.state.user = payload["sub"] # 사용자 정보 저장
except jwt.ExpiredSignatureError:
return HTTPException(status_code=401, detail="Token expired")
except jwt.InvalidTokenError:
return HTTPException(status_code=401, detail="Invalid token")
return await call_next(request)
# 인증된 사용자만 접근 가능한 API
@app.get("/protected")
async def protected_api(request: Request):
return {"message": f"Hello, {request.state.user}! This is a protected route."}
# 서버 실행: `uvicorn filename:app --reload`
- 직접지원은 하지않지만 Jinja2 템플릿 엔진을 이요해 렌더링
- pip install jinja2
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
app = FastAPI()
# Jinja2 템플릿 설정
templates = Jinja2Templates(directory="templates")
@app.get("/")
async def home(request: Request):
return templates.TemplateResponse("index.html", {"request": request, "message": "Hello, FastAPI!"})
Query()
사용 from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(q: str = Query("default_value")):
return {"q": q}