FastAPI - Plus

김기훈·2025년 10월 9일
FastAPI

FastAPI

목록 보기
5/7

테스트 스크립트

set -eo pipefail

COLOR_GREEN=`tput setaf 2;`
COLOR_NC=`tput sgr0;` # No Color

echo "Starting black"
poetry run black .
echo "OK"

echo "Starting ruff"
poetry run ruff check --select I --fix
poetry run ruff check --fix
echo "OK"

echo "Starting mypy"
poetry run mypy .
echo "OK"

echo "Starting pytest with coverage"
poetry run coverage run -m pytest
poetry run coverage report -m
poetry run coverage html
echo "OK"

echo "${COLOR_GREEN}All tests passed successfully!${COLOR_NC}"

# 생성 이후 chmod +x ./test.sh 를 사용해서 실행 권한을 추가

getattr , hasattr

  • getattr(object, name[, default])
    • object → 속성을 가져올 객체
    • name → 문자열로 된 속성 이름
    • default(선택) → 속성이 없을 때 대신 반환할 값 (없으면 AttributeError 발생)
class User:
    def __init__(self, username, age):
        self.username = username
        self.age = age

user = User("kihoon", 25)

print(getattr(user, "username"))  # "kihoon"
print(getattr(user, "age"))       # 25

print(getattr(user, "email", "no email"))  
# email 속성이 없으므로 "no email" 반환
  • hasattr(object, name) : 객체가 특정속성을 가지고 있는지 확인하는 내장 함수
    • object: 속성을 확인할 객체
    • name: 문자열로 된 속성 이름
      • 속성이 있으면 True , 없으면 Fasle
class User:
    def __init__(self, username):
        self.username = username

user = User("kihoon")

print(hasattr(user, "username"))  # True (username 속성이 존재)
print(hasattr(user, "age"))       # False (age 속성이 없음)
  • setattr(object, name, value) : 객체의 속성(변수)을 동적으로 설정할때 사용
    • 속성 설정/추가
    • object: 속성을 설정할 대상 객체
    • name: 속성 이름 (문자열)
    • value: 속성에 넣을 값
class User:
    pass

user = User()

# 속성 동적 추가
setattr(user, "username", "kim")
setattr(user, "age", 25)

print(user.username)  # kim
print(user.age)       # 25

username: str | None = None

  • username: str | None
    • username 변수는 문자열(str) 이거나 아예 값이 없는 상태(None)일 수 있다는 뜻
      • Union[str, None]과 같음
  • = None
    • 기본값을 None으로 지정
    • 이 변수를 꼭 넣지 않아도 되고, 값을 안 주면 자동으로 None이 들어감

items

  • kwargs.items() : 딕셔너리의 (키, 값) 쌍을 반환
kwargs = {"username": "kim", "age": 25}

for key, value in kwargs.items():
    print(key, value)
  • 1번째 반복 : username kim 출력
    • key = "username"
    • value = "kim"
  • 2번째 반복 : age 25 출력
    • key = "age"
    • value = 25

@classmethod

  • @classmethod에서는 첫 번째 인자로 cls를 받는데, 이건 메서드를 호출한 클래스 자신을 의미
class UserModel:
    _data = []        # 클래스 변수
    _id_counter = 1   # 클래스 변수

    @classmethod
    def get(cls, **kwargs):
        for user in cls._data:
            ...
  • cls는 UserModel 그 자체(클래스)를 가리킴
    • 따라서 cls._data는 사실상 UserModel._data와 똑같음

상속의 경우

class UserModel:
    _data = []
    @classmethod
    def get(cls):
        return cls._data   # 호출한 클래스의 _data

class AdminUserModel(UserModel):
    _data = ["admin"]  # 별도의 데이터 리스트

print(UserModel.get())      # []
print(AdminUserModel.get()) # ["admin"]
  • UserModel을 상속한 AdminUserModel 같은 클래스가 있다면, cls는 AdminUserModel이 됨

유효성 검사 옵션

옵션의미예시 코드허용되는 값
gt=nn보다 커야 함 (greater than)user_id: int = Path(gt=0)1, 2, 3, ...
ge=nn 이상 (greater or equal)age: int = Query(ge=18)18, 19, 20, ...
lt=nn보다 작아야 함 (less than)score: int = Query(lt=100)..., 97, 98, 99
le=nn 이하 (less or equal)level: int = Query(le=10)0, 1, 2, ..., 10
from fastapi import FastAPI, Path, Query

app = FastAPI()

@app.get("/users/{user_id}")
def read_user(
    user_id: int = Path(gt=0),               # 0보다 큰 값만 허용
    age: int = Query(ge=18),                 # 18세 이상만 허용
    score: int = Query(lt=100),              # 100 미만만 허용
    level: int = Query(le=10)                # 10 이하만 허용
):
    return {"user_id": user_id, "age": age, "score": score, "level": level}

# 경로 매개변수 (Path Parameter) = Path
                          @app.get("/users/{user_id}")
                          def read_user(user_id: int = Path(gt=0)):
                              return user_id

# 쿼리 매개변수 (Query Parameter) = Query
                        @app.get("/search")
                        def search(q: str = Query(min_length=3, max_length=10)):
                            return {"query": q}

# 본문 매개변수 (Request Body) = Body(Field)
# Pydantic 모델과 함께 사용하면 유효성 검사가 자동으로 적용
            from pydantic import BaseModel, Field

            class User(BaseModel):
                username: str = Field(min_length=3, max_length=20, regex="^[a-zA-Z0-9_]+$")
                age: int = Field(ge=18, le=100)

            @app.post("/users")
            def create_user(user: User):
                return user

async

async def는 "이 함수는 비동기 함수(Asynchronous Function)" 라는 선언

구분def (동기 함수)async def (비동기 함수)
실행 방식순차적으로 한 줄씩 실행됨여러 작업을 동시에(병렬처럼) 진행 가능
CPU 점유한 작업이 끝나야 다음 실행I/O 대기 시간 동안 다른 작업 수행 가능
반환값일반 값 (ex: int, str, dict)Coroutine 객체 반환
호출 방법그냥 func()await func() 로 호출해야 함
사용 예단순 계산, 로컬 처리네트워크 요청, DB 조회, 파일 I/O 등 느린 작업
  • def는 한개의 작업이 끝나야 다음 작업이 실행 가능한데
  • async def는 동시의 여러개의 작업이 가능함 (병렬느낌)
  • FastAPI는 내부적으로 비동기 서버(ASGI) 기반
    • DB, API, 파일 입출력 같은 I/O 작업이 많은 웹 서버에서 async def를 쓰면
      • CPU를 낭비하지 않고 동시에 여러 요청을 처리할 수 있음
from fastapi import FastAPI

app = FastAPI()

@app.get("/data")
async def get_data(): 
    await asyncio.sleep(3)   # 한 요청이 3초 동안 기다려도
    					     # 다른클라이언트 요청은 동시 처리 가능
    return {"message": "3초 후 응답 완료"}
  • 비동기 함수를 실행하려면 await를 쓰거나, asyncio.run() 같은 함수를 사용해야 한다
    • import asyncio 필요
import asyncio

async def hello():
    print("안녕!")
    await asyncio.sleep(1)
    print("잘가!")

asyncio.run(hello())  # asyncio를 사용해서 실행
  • 예외적으로 FastAPI 같은 프레임워크에서는 asyncio 안 써도 됨
    • FastAPI 내부에는 이미 이벤트 루프가 돌아가고 있어서 개발자가 직접 asyncio.run()을 쓸 필요가 없음
  • 하지만 “FastAPI는 asyncio 위에서 동작하지만, asyncio를 직접 쓴다면 import asyncio는 해야 한다.(ex: await asyncio.sleep(3))
from fastapi import FastAPI

app = FastAPI()

@app.get("/test")
async def test():
    return {"msg": "async 함수지만 asyncio import 필요 없음"}

# FastAPI 서버가 내부적으로 알아서 await를 처리

model_config

  • Pydantic v2에서 이전의 방식 class Config: 대신 모델의 설정(config) 을 지정하는 방법
  • model_config = {"extra": "forbid"}
    • 정의되지 않은 필드(예상치 못한 값)가 들어오면 에러를 내라
설정의미사용 상황
ignore정의되지 않은 필드는 무시가벼운 데이터 처리 시
allow정의되지 않은 필드도 허용유연한 입력 허용 시
forbid정의되지 않은 필드면 에러 발생 ⚠️API 검증을 엄격히 할 때 (✅ 추천)
class User(BaseModel):
    model_config = {"extra": "forbid"}
    model_config = {"extra": "allow"}
    model_config = {"extra": "ignore"}
    username: str

# {"username": "kim", "age": 25"} -> 오류 발생 
# {"username": "kim", "age": 25"} -> 정상 통과 (age는 내부 데이터에 남음)
# {"username": "kim", "age": 25"} -> 정상 통과 (age는 버려짐)

model_dump()

model_dump()는 Pydantic v2에서 새로 도입된 인스턴스 메서드

  • BaseModel을 상속한 객체(모델 인스턴스)에 대해 사용할 수 있다.
    • 이 메서드는 모델 객체를 딕셔너리로 변환
  • model_dump()는 단순히 딕셔너리로 바꾸는 것뿐 아니라
    • exclude_none=True : None값 제외 (user.model_dump(exclude_none=True))
    • include={'username'} : 특정 필드만 포함
    • exclude={'password'} : 특정 필드 제외
@app.post("/users")
def create_user(data: UserCreateRequest):
    # data는 UserCreateRequest(BaseModel 상속) 인스턴스
    user = UserModel.create(**data.model_dump())
    return user.id
  • data.model_dump()
    • 요청으로 받은 UserCreateRequest 객체를 딕셔너리로 변환해서 create() 함수의 인자로 넘기기 위한 것
구분Pydantic v1Pydantic v2설명
딕셔너리 변환model.dict()model.model_dump()모델 인스턴스를 dict로 변환
JSON 변환model.json()model.model_dump_json()JSON 문자열로 변환

async 와 await의 기본 관계

  • async: 비동기 함수(코루틴) 정의 = “비동기로 실행될 수 있는 함수”를 만든다.
  • await: 비동기 작업의 결과를 기다림 = “그 함수가 끝날 때까지 잠시 멈춘다”

동기 비동기 코드 비교

# 동기 - 물이 끓을 동안 아무것도 못 함 (CPU 대기)
def make_ramen():
    boil_water()     # 물 끓을 때까지 기다림
    put_noodle()     # 라면 넣기
    eat()            # 먹기


# 비동기 - await는 “지금 이 작업 끝날 때까지, 다른 일 먼저 하자” 라는 의미
# CPU가 놀지 않고 다른 비동기 작업을 실행할 수 있게 해주는 키워드
async def make_ramen():
    await boil_water()   # 물이 끓는 동안 잠깐 멈추고
    await put_noodle()   # 라면 넣기
    await eat()          # 먹기

의존성

  • “경로 함수(엔드포인트)가 동작하는 데 필요한 외부의 로직이나 자원을 주입받는 방식”
    • “경로 함수에서 필요로 하는 공통 자원을 외부 함수로 분리하고,
      • FastAPI가 알아서 실행하고 주입해주는 구조”
역할설명예시
공통 로직 분리인증, DB 연결, 설정 로드 등 공통 부분을 재사용get_current_user, get_db
자동 주입FastAPI가 알아서 실행 순서를 조정하고 값을 주입Depends(get_db)
테스트 용이성테스트 시 mock 객체를 넣기 쉬움유닛 테스트 시 DB 대체
가독성 향상경로 함수가 “핵심 기능”만 담당인증/검증은 분리

예제

  • 예제 1 — 인증 처리
from fastapi import Depends, HTTPException

def get_current_user(token: str = "fake-token"):
    if token != "fake-token":
        raise HTTPException(status_code=403, detail="Invalid token")
    return {"username": "kihoon"}

@app.get("/users/me")
async def read_user(current_user: dict = Depends(get_current_user)): ⭐️
    return current_user

  • read_user(): Depends() 덕분에 직접 인증을 하지 않아도 됨.

    • FastAPI가 자동으로 get_current_user()를 실행해 결과를 넣어줌

  • 예제 2 — 데이터베이스 연결 의존성

def get_db():
    db = connect_to_database()
    try:
        yield db  # 호출자에게 db 세션 제공
    finally:
        db.close()

@app.get("/items/")
async def get_items(db=Depends(get_db)): ⭐️
    return db.query("SELECT * FROM items")
  • yield는 요청 전후 처리를 함께 관리할 때 사용
    • “DB 세션 열기 → 요청 처리 → 세션 닫기” 구조

미들웨어

  • 요청(Request) → 미들웨어 ⭐️ → 엔드포인트(경로 함수) → 응답(Response)
    • 순서로 실행되는 중간 처리 계층
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http") ⭐️ # 변형하면 안됨 "고정"
async def my_middleware(request: Request, call_next):
    print("📥 요청이 들어옴:", request.url.path)

    response = await call_next(request)  # 다음 단계(엔드포인트)로 넘김

    print("📤 응답이 나감:", response.status_code)
    return response
  • request: 들어온 HTTP 요청 객체
    • call_next: 요청을 다음 단계로 넘겨주는 함수 (엔드포인트 실행 담당)
      • response: 실제 엔드포인트에서 반환된 응답 객체

예시

  • 요청 시간 측정 미들웨어
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.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(round(process_time, 3))
    return response
  • 결과적으로 클라이언트는 응답 헤더에서 처리 시간을 볼 수 있음: X-Process-Time: 0.005

JWT (JSON Web Token)

  • “사용자의 인증 정보를 안전하게 주고받기 위한 토큰 형식”
    • 즉, 서버가 인증된 사용자임을 증명하는 증명서 같은 것을 발급해 주는 것
      • 이후 사용자가 이걸 요청마다 들고다니면, 서버는 다시 로그인 검사를 안해도 됨

JWT 방식 (토큰 기반)

  • 로그인 성공 시 서버가 토큰을 발급
    • 클라이언트는 그걸 로컬스토리지 / 헤더에 저장
      • 이후 요청 시 헤더에 Authorization: Bearer <JWT> 붙여서 전송
        • 서버는 DB 안 보고 토큰 자체를 검증해서 신뢰
  • 장점
    • 서버가 “상태를 저장하지 않음”(stateless)
    • 확장성 높고 빠름

JWT 구조

  • 세 부분으로 나눠져 있으며 각 부분은 Base64로 인코딩되어 하나의 문자열로 이어짐
구분이름역할
1️⃣Header토큰 타입(JWT)과 해싱 알고리즘 정보 (HS256, RS256 등)
2️⃣Payload실제 데이터(사용자 ID, 만료 시간 등)
3️⃣Signature위 두 부분을 비밀키로 서명한 값 — 위조 방지용
Header:
{
  "alg": "HS256",
  "typ": "JWT"
}

Payload:
{
  "sub": "user123",
  "name": "Kihoon",
  "exp": 1739659223
}

# 결과
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.
eyJzdWIiOiJ1c2VyMTIzIiwibmFtZSI6IktoaG9vbiIsImV4cCI6MTczOTY1OTIyM30.
YH4I0B3gQG3xM7bPrlK6E4Zw5n6G7N-s6H0Md7K7owk

JWT 동작 흐름

  • 로그인 시
    • 클라이언트가 ID/PW로 로그인
    • 서버가 검증 후 JWT 생성 (access_token)
    • 클라이언트는 그걸 저장 (보통 Authorization 헤더나 로컬스토리지)
  • 인증이 필요한 경우
    • 클라이언트가 Authorization: Bearer <token> 붙여서 요청
    • 서버가 토큰을 복호화 및 검증
    • 정상 토큰이면 사용자 정보 추출 → 요청 수행

FastAPI 예시

from datetime import datetime, timedelta
from jose import JWTError, jwt

SECRET_KEY = "mysecretkey"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 토큰 생성
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 토큰 검증
def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

Annotated[X,Y]

  • “이 값의 타입은 X인데, 추가적인 정보 Y도 함께 붙인다” 라는 뜻
from typing import Annotated

Age = Annotated[int, "단위: 세"] ⭐️

def set_age(age: Age):
    print(age)
  • 이 코드는 사실상 def set_age(age: int) 과 똑같이 작동
    • “이 Age는 int이면서 설명 정보가 있다” 라는 메타데이터를 함께 기록
      • 메타데이터: API 전체(혹은 각 경로)에 대한 설명 정보를 담는 데이터
      • 메타데이터는 실제 API 동작에는 영향을 주지 않음
      • 자동 문서화(Swagger UI, ReDoc) 에서 설명을 표시하기 위해 사용

헤더(header)

입력 위치Swagger에서 표시예시
헤더(Header)“Headers” 영역X-Token: …
쿼리(Query)“Query params” 영역?q=…
본문(Body)“Request body” 영역JSON / form 등
  • Swagger에선 “사용자가 값을 넣는 입력창이 생긴다”는 점에서 비슷하게 보임
    • 하지만 FastAPI가 어디에서 값을 꺼내는지가 완전히 다르다.

일반 매개변수 / 헤더 매개변수 비교

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

# GET /items/?q=hello

@app.get("/items/")
async def read_items(x_token: Annotated[str, Header()]):
    return {"token": x_token}

# GET /items/
# X-Token: fake-token

@app.get("/check/")
async def check(
    q: str | None = None,
    x_token: Annotated[str | None, Header()] = None,
):
    return {"query": q, "header": x_token}

# GET /check?q=python
# {"query": "python", "header": null}

# GET /check
# X-Token: secret
# {"query": null, "header": "secret"}
  • q 값은 URL의 쿼리스트링(?q=hello)에서 가져옴.
  • x_token 값은 HTTP 헤더(X-Token) 에서 가져옴.

헤더도 사람이 입력하는 값이고 중요한점은 “입력 위치(어디에 넣느냐)” 와 “의미(무엇을 위한 데이터냐)” 가 다르다는 것

왜 이렇게 나뉘어져 있을까

구분위치용도예시
쿼리(Query)URL ?q=value검색/필터/옵션/search?q=python
본문(Body)요청 내용(JSON, form 등)실제 데이터{ "username": "kihoon" }
헤더(Header)요청 메타정보인증, 형식 등Authorization: Bearer abc
위치예시목적FastAPI에서 사용하는 방식
쿼리(Query)/items?q=apple요청 데이터나 검색 조건q: str or Query()
본문(Body){ "username": "kihoon" }실제 리소스 데이터 (POST/PUT 등)user: UserSchema or Body()
헤더(Header)X-Token: abc123인증, 형식, 언어, 클라이언트 정보 등x_token: Header()
profile
김기훈
안녕하세요.
이전 포스트

FastAPI 용어 정리

다음 포스트

FastAPI-JWT

0개의 댓글