FastAPI 기본

Haks.·2025년 2월 13일
0

How to use

목록 보기
25/32

FastAPI

  • Python으로 RESTful API를 쉽고 빠르게 개발하는 모던 웹 프레임워크
  • 비동기 지원
  • 최소한의 코드로 동작하는 API
  • 고성능 Starlette, Pydantic 기반으로 높은 처리 속도 제공
  • 강력한 타입검사 : Python 타입 힌트를 적극 활용
  • 자동화된 문서화 : OpenAPI 명세와 Swagger UI를 자동 생성

주요 사례

  • 웹 애플리케이션 : CRUD API, 대시보드, 비즈니스 로직 처리
  • 마이크로 서비스 : 비동기와 경량 설계로 효율적인 마이크로서비스 작성
  • 머신러닝/AI 배포 : FastAPI를 통해 AI모델을 웹 API로 간단히 배포

특징

  • 비동기 처리(Asynchronous Programming)

    • 멀티 태스크, async/await를 기본적으로 지원: 높은 요청 처리량 보장
    • 동시 요청 처리에 유리: DB쿼리, 외부 API 호출 같은 I/O 작업에서 효율적
    • 동시요청 처리에 유리
  • 데이터 검증 및 직렬화

    • Pydantic 모델을 통해 데이터 검증과 직렬화를 자동 처리
    • 클라이언트 요청 (Request) 데이터를 Python 객체로 변환 + 응답 데이터를 JSON으로 변환
  • 경량설계

    • 필요한 것만 추가해서 사용
  • 타입 힌트 기반 개발

    • python 타입 힌트를 활용하여 개발자가 타입 오류를 줄이고 유비보수 쉽게 가능

기본 코드 구조

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}

FastAPI 경로 매개변수 (Path Parameters)

  • 경로 매개변수란?
    • URL 경로의 일부로 동적인 값을 전달
    • 예: /items/9에서 9는 동적인 값
@app.get("/items/{item_id}")
def read_item(item_id: int):
	return {"item_id" :item_id}

FastAPI 쿼리 매개변수 (Query Parameters)

  • 쿼리 매개변수란?
    • URL에서 ?key=value 형식으로 전달되는 값
    • 경로 매개변수와 달리 선택적으로 사용할 수 있음

/items/?skip=5&limit=15

@app.get("/items/")
def read_item(skip: int=0, limit: int = 10):
	return {"skip": skip, "limit": limit}

FAST 비동기 처리(Asynchronous)

  • 비동기의 개념
    • 동기: 순차적
    • 비동기: 작업이 비차단적 같이 실행

await

  • 비동기 함수 내에서 순차적으로 실행하지만
  • 같은 비동기 함수가 asyncio.gather() 또는 asyncio.create_task()로 묵여서 실행되었을때 순차적으로 진행되는걸 지정
  • await 함수가 실행중이면 그 밑의 작업은 진행하지 않음, 같이 실행된 다른 비동기 함수는 동작중
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()가 이후에 종료됨)

Templates

구조, 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

  • Pydantic은 FastAPI에서 데이터 검증과 직렬화(Serialization)를 쉽게 처리할 수 있도록 도와주는 라이브러리
  • Python의 데이터 유효성 검사 및 데이터 변환을 쉽게 처리
  • FastAPI의 요청(Request) 및 응답(Response) 데이터를 검증할 때 필수적으로 사용됨
  • 타입 힌트를 기반으로 데이터 검증을 자동으로 수행!
  • pydantic, BaseModel
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} 

Field 사용법

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

Optional

  • 선택적 값, 기본값이 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 요청에서 여러 개의 값을 받을 때
    • Query를 사용해야 데이터가 제대로 fetch됨
    • [GET] /items/?names=hello&names=world
def read_items(names: List[str] =Query(default=[])):
  • Dict[str, T]: JSON 데이터를 받을 때
    • Query를 사용해야 데이터가 제대로 fetch됨
    • [GET] /config/?settings={"theme": "light"}
def get_config(settings: str = Query('{theme":"dark"}')): # Json문자열로 받음

Validator

  • 구조

    • 검증 함수는 항상 클래스메서드,
    • 검증 실패시 반드시 에러를 발생시켜야함, ValueError
  • 동작

    • 모델 초기화 시, 해당 필드의 값이 검증 로직을 통과해야함
    • 통과하지 못하면 예외 메세지가 반환됨

Swagger UI & ReDoc이란?

  • Swagger UI : /docs
  • ReDoc : /redoc
  • Swagger UI
    • FastAPI에서 자동 생성되는 API 문서
    • OpenAPI 명세 기반으로 작동
    • 주요 기능
      • 시각화: 모든 API 엔드포인트와 요청/응답을 쉽게 확인
      • 테스트 가능: API 요청을 직접 실행하고 응답 확인
  • ReDoc
    • Swagger UI의 대안으로 제공되는 API 문서화 도구

OpenAPI 명세와 메타데이터 설정

  • OpenAPI 명세란?
    • RESTfual API의 정의를 표준화한 문서 형식
    • API의 엔드포인트, 요청/응답 데이터, 상태 코드, 인증방식을 정의
    • JSON 또는 YAML 형식으로 작성
  • Swagger UI or Redoc 으로 명세
  • OpenAPI 메타데이터 추가
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"])

MiddleWare

요청과 응답 사이에서 실행되는 함수, 모든 요청이 들어오거나 응답이 나나기 전에 실행되는 중간 처리 단계

사용법

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())로 요청이 전달됨.

  • jwt토큰 예제
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`

FastAPI HTML 렌더링

  • 직접지원은 하지않지만 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!"})

Tip

  • 경로 설정시 우선순위 생각해야함
    • 같은경로상에 찾는거 잘생각하자 /{id} 와/serarch 같으면 search가 id로 들어가서 찾을수도 있음
  • CSR: 클라이언트 사이드 렌더링은? : 클라이언트 사이드 렌더링(CSR)은 웹 애플리케이션에서 초기 HTML을 서버에서 최소한으로 제공하고, JavaScript가 브라우저에서 실행되어 동적으로 페이지를 구성하는 방식이야.
  • 쿼리 매개변수 기본값 Query() 사용
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: str = Query("default_value")):
    return {"q": q}

0개의 댓글

관련 채용 정보