Study/fastapi

[udemy] Complete FastAPI masterclass from scratch - 학습 정리 1

bluebamus 2025. 1. 27.

https://www.udemy.com/course/completefastapi/?couponCode=KEEPLEARNING

 

- 학습 중 필요한 내용을 정리하거나, 추가로 더 살펴본 학습에 대한 정리만 함

 

1. GET method

   - Predefined values : 패스 파라미터의 값을 Enum을 이용해 특정 범주로 제한할 수 있다.

      - /docs로 접속하여 swagger를 사용하면 type에 열거형으로 펼쳐지는 선택 항목들을 볼 수 있다.

from enum import Enum

class BlogType(str, Enum):
    short = 'short'
    story = 'story'
    howto = 'howto'

@app.get('/blog/type/{type}')
def get_blog_type(type: BlogType):
  return {'message': f'Blog type {type}'}

 

2. Operation description

   1) Summary and Description and Response description

      - swagger 문서에 반영되는 항목들을 엔드포인트별로 정의할 수 있다.

@app.get(
  '/blog/all',
  tags=['blog'],
    summary='Retrieve all blogs',
    description='This api call simulates fetching all blogs',
    response_description="The list of available blogs"
  )
def get_blogs(page = 1, page_size: Optional[int] = None):
  return {'message': f'All {page_size} blogs on page {page}'}

 

3. Operation description

   1) Routers

      - routers를 사용하면, 같은 파일의 엔드포인트들은 동일한 url prefix와 tags를 정의할 수 있다. 

         - 엔드포인트에 직접 작성하던 반복 작업들을 router로 줄일 수 있다. 

         - url prefix 기준으로 여러 파일로 프로젝트를 분리, 관리할 수 있다.

 

4. Parameters

   1) parameter metadata

      - path()와 query() 메타 데이터 정리

필드 설명 path() 예제 query() 예제
default 기본값 path(default=42) query(default="default_value")
alias 매개변수의 별칭 path(..., alias="item_id") query(..., alias="search_term")
title 매개변수 제목 path(..., title="Item ID") query(..., title="Search term")
description 매개변수 설명 path(..., description="The ID of the item.") query(..., description="Term to search for.")
example 단일 예제 path(..., example=123) query(..., example="fastapi")
examples 여러 예제 path(..., examples={"example1": {"value": 1}}) query(..., examples={"example1": {"value": "test"}})
ge 최소값 (포함) path(..., ge=1) query(..., ge=10)
gt 최소값 (초과) path(..., gt=0) query(..., gt=5)
le 최대값 (포함) path(..., le=100) query(..., le=50)
lt 최대값 (미만) path(..., lt=10) query(..., lt=20)
min_length 문자열 최소 길이 path(..., min_length=3) query(..., min_length=5)
max_length 문자열 최대 길이 path(..., max_length=50) query(..., max_length=100)
regex 문자열 정규 표현식 패턴 path(..., regex="^item_[0-9]+$") query(..., regex="^query_[a-z]+$")
deprecated 매개변수 사용 중단 여부 표시 - query(..., deprecated=True)

 

      - 예시 코드

from fastapi import FastAPI, Path, Query

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(
        ..., 
        title="Item ID", 
        ge=1, 
        le=1000, 
        description="The ID of the item to retrieve.",
        example=42
    ),
    q: str = Query(
        None, 
        title="Query string", 
        max_length=50, 
        description="Search query for items.", 
        regex="^query_[a-z]+$", 
        example="query_test"
    )
):
    return {"item_id": item_id, "q": q}

 

   2) Body()란?

      - Body()는 FastAPI의 의존성 함수 중 하나로, HTTP 요청 본문에서 데이터를 받아오기 위해 사용된다. 이를 통해 요청 본문에 포함된 JSON 데이터 등의 구조를 Pydantic 모델이나 기본 Python 데이터 타입으로 매핑할 수 있다.

         1. Pydantic 모델 통합: Body()는 Pydantic 모델과 함께 사용하여 데이터 검증과 변환을 자동화한다.
         2. 메타데이터 지원: 문서화를 위한 제목, 설명, 예제 등의 메타데이터를 제공한다.
         3. JSON 요청 처리: 기본적으로 JSON 요청 본문을 처리하며, 데이터 타입과 형식을 자동으로 매핑한다.
         4. embed 옵션: 요청 본문을 루트 키 없이 JSON으로 처리할 수 있다.

 

      -  Body()의 주요 메타데이터

         - min_length 혹은 max_length와 같은 조건 설정도 가능함

         - 정규식 표현을 위한 regex을 이용한 조건 설정도 가능함

필드 설명 예제
default 요청 본문의 기본값. 생략할 경우 요청 본문은 필수 값이 됨. Body(default={"key": "value"})
embed 요청 본문을 "key"로 감싸지 않고 그대로 JSON 루트로 처리할지 여부. 기본값은 False. Body(..., embed=True)
title 요청 본문 필드의 제목. OpenAPI 문서에 표시됨. Body(..., title="User Info")
description 요청 본문 필드의 설명. OpenAPI 문서에 표시됨. Body(..., description="Information about the user.")
example 요청 본문의 단일 예제. OpenAPI 문서에서 사용할 수 있음. Body(..., example={"name": "John", "age": 30})
examples 요청 본문의 여러 예제를 제공. OpenAPI 문서에서 사용할 수 있음. Body(..., examples={"example1": {"value": {"name": "Alice"}}})

 

      - 기본 사용법

         - Body()를 사용하여 요청 본문에서 데이터를 받아오는 기본 예제

from fastapi import FastAPI
from pydantic import BaseModel
from fastapi import Body

app = FastAPI()

class User(BaseModel):
    name: str
    age: int

@app.post("/users/")
def create_user(user: User = Body(...)):
    return {"user": user}

 

      - 요청 예시 (POST /users/)

{
  "name": "John",
  "age": 30
}

 

      - 응답 예시

{
  "user": {
    "name": "John",
    "age": 30
  }
}

 

      - Body 메타데이터 활용

from fastapi import FastAPI, Body
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    description: str = None

@app.post("/items/")
def create_item(
    item: Item = Body(
        ...,
        title="Item details",
        description="Details of the item to create",
        example={"name": "Laptop", "price": 1500.0, "description": "High-end gaming laptop"}
    )
):
    return {"item": item}

 

      - 요청 예시 (POST /items/)

{
  "name": "Laptop",
  "price": 1500.0,
  "description": "High-end gaming laptop"
}

 

      - 응답 예시

{
  "item": {
    "name": "Laptop",
    "price": 1500.0,
    "description": "High-end gaming laptop"
  }
}

 

      - embed 활용

         - embed=True를 사용하면 요청 본문이 특정 키로 감싸지지 않고, 루트 JSON으로 직접 데이터를 받을 수 있다.

from fastapi import FastAPI, Body

app = FastAPI()

@app.post("/embed/")
def create_item(data: dict = Body(..., embed=True)):
    return {"data": data}

 

      - 요청 예시 (POST /embed/)

{
  "key": "value"
}

 

      - 응답 예시

{
  "data": {
    "key": "value"
  }
}

 

      - examples 필드 사용

         - 여러 요청 본문 예제를 문서화에 포함할 수 있다.

from fastapi import FastAPI, Body
from pydantic import BaseModel

app = FastAPI()

class Product(BaseModel):
    name: str
    price: float

@app.post("/products/")
def create_product(
    product: Product = Body(
        ...,
        examples={
            "basic": {
                "summary": "Basic example",
                "description": "A simple example of a product.",
                "value": {"name": "Book", "price": 10.99},
            },
            "premium": {
                "summary": "Premium example",
                "description": "A premium product example.",
                "value": {"name": "Laptop", "price": 1500.00},
            },
        }
    )
):
    return {"product": product}

 

      - min_length

         - 문자열의 최소 길이를 제한

from fastapi import FastAPI, Body

app = FastAPI()

@app.post("/items/")
def create_item(
    description: str = Body(..., min_length=10)
):
    return {"description": description}

 

      - max_length

         - 문자열의 최대 길이를 제한

from fastapi import FastAPI, Body

app = FastAPI()

@app.post("/items/")
def create_item(
    description: str = Body(..., max_length=50)
):
    return {"description": description}

 

      - regex

         - 문자열의 형식과 패턴을 정규 표현식으로 정의

from fastapi import FastAPI, Body

app = FastAPI()

@app.post("/items/")
def create_item(
    description: str = Body(..., regex="^item_[0-9]+$")
):
    return {"description": description}

 

   3) FastAPI Query 파라미터: 다중 값 처리 방법 정리

유형 설명 코드 예제 요청 예시 응답 예시
1. List 기본 처리 동일한 파라미터 이름으로 여러 값을 전달받아 list로 처리. def get_items(q: list[str] = Query(None)):
    return {"q": q}
/items/?q=apple&q=banana&q=cherry json { "q": ["apple", "banana", "cherry"] }
2. List 기본값 설정 요청이 없는 경우 기본값을 설정 가능. def get_items(q: list[str] = Query(["default1", "default2"])):
    return {"q": q}
/items/ json { "q": ["default1", "default2"] }
3. List 길이 제한 리스트의 최소/최대 길이를 제한 가능. def get_items(q: list[str] = Query(None, min_length=1, max_length=5)):
    return {"q": q}
/items/?q=apple&q=banana json { "q": ["apple", "banana"] }
4. Dict (JSON) 쿼리 파라미터로 JSON 문자열을 받아 dict로 변환. def get_dict(data: str = Query(...)):
    return {"data": json.loads(data)}
/dict/?data={"key1":"value1","key2":"value2"} json { "data": { "key1": "value1", "key2": "value2" } }
5. Dict (Pydantic) Pydantic 모델을 활용해 딕셔너리 형태의 데이터를 처리. class Data(BaseModel):
    key1: str
    key2: str

@app.get("/dict/")
def get_dict(data: Data):
    return {"data": data}
/dict/?key1=value1&key2=value2 json { "data": { "key1": "value1", "key2": "value2" } }
6. List+Dict (JSON) 리스트 내에 딕셔너리 구조를 JSON 문자열로 받아 처리. def get_list_dict(data: str = Query(...)):
    return {"data": json.loads(data)}
/list-dict/?data=[{"key1":"value1"},{"key2":"value2"}] json { "data": [ { "key1": "value1" }, { "key2": "value2" } ] }
7. List+Dict (Pydantic) 리스트와 딕셔너리 구조를 Pydantic 모델로 처리. class Item(BaseModel):
    key: str
    value: str

@app.get("/list-dict/")
def get_items(data: List[Item]):
    return {"data": data}
/list-dict/?data=[{"key":"key1","value":"value1"},{"key":"key2","value":"value2"}] json { "data": [ { "key": "key1", "value": "value1" }, { "key": "key2", "value": "value2" } ] }
8. Tuple 처리 튜플 데이터를 다중 파라미터로 받아 처리. def get_tuple(data: Tuple[int, int, int] = Query(...)):
    return {"data": data}
/tuple/?data=1&data=2&data=3 json { "data": [1, 2, 3] }
9. Pydantic 필드 추가 Pydantic 필드를 사용해 더 복잡한 구조의 데이터 처리. def get_items(data: List[Item]):
    return {"data": data}
/complex/?data=[{"key":"key1","value":"value1"},{"key":"key2","value":"value2"}] json { "data": [ { "key": "key1", "value": "value1" }, { "key": "key2", "value": "value2" } ] }

 

   4) FastAPI의 데이터 유효성 검사 옵션

검증 파라미터 설명 예시
gt Greater Than: 값이 지정된 값보다 커야 한다. gt=10 (10보다 큰 값)
ge Greater Than or Equal: 값이 지정된 값 이상이어야 한다. ge=10 (10 이상)
lt Less Than: 값이 지정된 값보다 작아야 한다. lt=10 (10보다 작은 값)
le Less Than or Equal: 값이 지정된 값 이하이어야 한다. le=10 (10 이하)
max_length 문자열 길이가 지정된 최대값을 넘지 않도록 제한한다. max_length=100 (100자 이하)
min_length 문자열 길이가 지정된 최소값 이상이어야 한다. min_length=5 (5자 이상)
regex 정규식을 사용하여 값이 특정 패턴을 따르도록 제한한다. regex="^[A-Za-z0-9]+$" (영문자와 숫자만 허용)
multiple_of 값이 지정된 값으로 나누어 떨어져야 한다. multiple_of=5 (5의 배수)
max_items 리스트의 항목 수가 지정된 최대값을 넘지 않도록 제한한다. max_items=10 (10개 이하)
min_items 리스트의 항목 수가 지정된 최소값 이상이어야 한다. min_items=1 (1개 이상)
from fastapi import FastAPI
from pydantic import BaseModel
from pydantic import conint, constr
from typing import List

app = FastAPI()

# Path 파라미터 예시
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# Query 파라미터 예시: gt, ge, lt, le
@app.get("/items/")
async def read_item_price(price: conint(ge=10, le=100)):
    return {"price": price}

# Body 파라미터 예시: max_length, min_length, regex
class Item(BaseModel):
    name: constr(min_length=3, max_length=50)
    description: str

@app.post("/items/")
async def create_item(item: Item):
    return {"name": item.name, "description": item.description}

# 리스트 파라미터 예시: min_items, max_items
@app.post("/items/")
async def create_items(items: List[int] = Query(..., min_items=1, max_items=5)):
    return {"items": items}

 

5. Database with SQLAlchemy

   1) ORM 사용시, pydactic을 database와 연동하는 방법

      - 기본적으로 Pydantic은 데이터가 딕셔너리 형식일 때만 모델을 직렬화하고 변환한다. 그러나 SQLAlchemy 객체는 딕셔너리가 아니기 때문에 orm_mode를 설정해야 이를 Pydantic 모델로 변환할 수 있다.

class UserDisplay(BaseModel):
  username: str
  email: str
  class Config():
    orm_mode = True
from typing import List
from schemas import UserBase, UserDisplay
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from db.database import get_db
from db import db_user


router = APIRouter(
  prefix='/user',
  tags=['user']
)

# Create user
@router.post('/', response_model=UserDisplay)
def create_user(request: UserBase, db: Session = Depends(get_db)):
  return db_user.create_user(db, request)

# Read all users
@router.get('/', response_model=List[UserDisplay])
def get_all_users(db: Session = Depends(get_db)):
  return db_user.get_all_users(db)

# Read one user
@router.get('/{id}', response_model=UserDisplay)
def get_user(id: int, db: Session = Depends(get_db)):
  return db_user.get_user(db, id)

 

   2) update시 orm을 이용하는 방법

      - django와 달리, 대상이 되는 필드를 "DB 클래스.필드명"으로 정의해야 한다.

from .database import Base
from sqlalchemy import Column, Integer, String


class DbUser(Base):
  __tablename__ = 'users'
  id = Column(Integer, primary_key=True, index=True)
  username = Column(String)
  email = Column(String)
  password = Column(String)
def update_user(db: Session, id: int, request: UserBase):
  user = db.query(DbUser).filter(DbUser.id == id)
  user.update({
    DbUser.username: request.username,
    DbUser.email: request.email,
    DbUser.password: Hash.bcrypt(request.password)
  })
  db.commit()
  return 'ok'
# Update user
@router.post('/{id}/update')
def update_user(id: int, request: UserBase, db: Session = Depends(get_db)):
  return db_user.update_user(db, id, request)

6. Concepts

   1) Error handling

      - 예외처리로 HTTPException을 사용할 수 있고 status_code로 fastapi에서 제공하는 status를 사용할 수 있다.

from fastapi import HTTPException, status

def get_article(db: Session, id: int):
  article = db.query(DbArticle).filter(DbArticle.id == id).first()
  if not article:
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
      detail=f'Article with id {id} not found')
  return article

 

      1. FastAPI에서 커스텀 Exception 정의 및 사용 방법

         - 사용 방법 :

            - python의 기본 Exception 클래스를 상속받아 정의한다.

class CustomException(Exception):
    def __init__(self, name: str, detail: str):
        self.name = name
        self.detail = detail

 

            - FastAPI의 Exception 처리기 생성

               - FastAPI의 @app.exception_handler 데코레이터를 사용하여 커스텀 Exception에 대한 처리기를 정의한다.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=400,
        content={
            "message": f"Custom Exception occurred: {exc.name}",
            "detail": exc.detail,
        },
    )

 

            - 커스텀 Exception 사용

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 1:
        raise CustomException(name="InvalidItemID", detail="Item ID must be greater than 0")
    return {"item_id": item_id}

 

      2. HTTPException 오버라이딩

         - FastAPI의 기본 HTTPException을 오버라이딩하면 좀 더 세부적으로 커스터마이징된 HTTP 에러 처리를 구현할 수 있다.

         - HTTPException 오버라이딩

from fastapi.exceptions import HTTPException

class CustomHTTPException(HTTPException):
    def __init__(self, status_code: int, detail: str, code: str = "error"):
        super().__init__(status_code=status_code, detail=detail)
        self.code = code

 

         - HTTPException 처리기 커스터마이징

@app.exception_handler(CustomHTTPException)
async def custom_http_exception_handler(request: Request, exc: CustomHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error_code": exc.code,
            "detail": exc.detail,
        },
    )

 

         - 오버라이딩된 HTTPException 사용

@app.get("/custom-error")
async def custom_error():
    raise CustomHTTPException(
        status_code=404,
        detail="Custom HTTP Exception Example",
        code="NOT_FOUND"
    )

 

   2) Custom Response

      - 응답에 파일이나 xml 등 여러가지의 부가적인 데이터를 전달해야 하는 경우가 있다. 이런 경우 custom response를 사용할 수 있다. 

products = ['watch', 'camera', 'phone']

@router.get('/all')
def get_all_products():
  # return products
  data = " ".join(products)
  return Response(content=data, media_type="text/plain")

 

      - 오류가 있을 경우 일반 text로 응답으로 보내고, 정상적인 경우 html 문서를 보낼 경우, swagger 문서에서도 이러한 정보가 필요할 수 있다. endpoint를 정의하는 데코레이터에 responses={}를 이용해 정의할 수 있다.

products = ['watch', 'camera', 'phone']

@router.get('/{id}', responses={
  200: {
    "content": {
      "text/html": {
        "example": "<div>Product</div>"
      }
    },
    "description": "Returns the HTML for an object"
  },
  404: {
    "content": {
      "text/plain": {
        "example": "Product not available"
      }
    },
    "description": "A cleartext error message"
  }
})
def get_product(id: int):
  if id > len(products):
    out = "Product not available"
    return PlainTextResponse(status_code=404, content=out, media_type="text/plain")
  else:
    product = products[id]
    out = f"""
    <head>
      <style>
      .product {{
        width: 500px;
        height: 30px;
        border: 2px inset green;
        background-color: lightblue;
        text-align: center;
      }}
      </style>
    </head>
    <div class="product">{product}</div>
    """
    return HTMLResponse(content=out, media_type="text/html")

 

   3) Headers

      - 입력 인자의 타입 정의에서 Header()를 사용하여 헤더 정보를 가져올 수 있다. 해당 정보는 client에서 헤더에 custom header정보를 전달할 경우 사용할 수 있다.

         - 중요한 접으로 python에서는 인자 값에 '-' 표시를 할 수 없다 때문에 '_'를 사용해 표시한다. 하지만 식별해야 하는 헤더의 이름에 실재로는 '-'가 적용되어 찾게된다.

# 하나의 값만 전달할 경우
def get_products(
  response: Response,
  custom_header: Optional[str] = Header(None)
  ):

# List를 이용해 여러개의 값을 전달할 경우
def get_products(
  response: Response,
  custom_header: Optional[List[str]] = Header(None)
  ):

 

      - 서버에서 custom 헤더를 보내는 방법으로 response.headers['name'] = "" 를 사용하면 된다.

@router.get('/withheader')
def get_products(
  response: Response,
  custom_header: Optional[List[str]] = Header(None)
  ):
  if custom_header:
    response.headers['custom_response_header'] = " and ".join(custom_header)
  return {
    'data': products,
    'custom_header': custom_header
  }

 

   4) Cookies

      - 쿠키를 사용하는 방법으로 .set_cookie를 이용해 정의하고, 입력 인자에 Cookie(None)을 사용해 값을 가져오는 방법이 있다.

      - 쿠키 설정 :

def get_all_products():
  # return products
  data = " ".join(products)
  response = Response(content=data, media_type="text/plain")
  response.set_cookie(key="test_cookie", value="test_cookie_value")
  return response

 

      - 쿠키 입력 인자로 가져오기

def get_products(
  response: Response,
  custom_header: Optional[List[str]] = Header(None),
  test_cookie: Optional[str] = Cookie(None)
  ):
  ...

 

   5) Form data

      - 필수 라이브러리 설치

pip insetall python-multipart

 

      - 인자 값으로 Form() 사용

         - Form()은 Path(), Query() 등과 같이 입력 인자에 대한 다양한 옵션을 정의할 수 있다. 

from fastapi import APIRouter, Header, Cookie, Form

@router.post('/new')
def create_product(name: str = Form(...)):
  products.append(name)
  return products

 

   6) CORS

      - ip와 port 주소가 다른 서버에서 내 서버에 접속을 시도하는 경우 cors 설정이 되어 있지 않다면, 원칙적으로 에러가 발생한다. 웹 브라우저는 처음 접속한 서버의 정보를 바탕으로 이러한 에러를 발생하게 된다.

      - fastapi에서는 CORSMiddleware 미들웨어를 사용하여 관련한 설정을 쉽게 할 수 있다.

from fastapi.middleware.cors import CORSMiddleware

origins = [
  'http://localhost:3000'
]

app.add_middleware(
  CORSMiddleware,
  allow_origins = origins,
  allow_credentials = True,
  allow_methods = ["*"],
  allow_headers = ['*']
)

 

7. Authentication

   1) token 생성 endpoint 호출 및 JWT 토큰 생성 

      - JWT 라이브러리 설치 :

pip install python-jose
@router.post('/token')
def get_token(request: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
  user = db.query(models.DbUser).filter(models.DbUser.username == request.username).first()
  if not user:
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invalid credentials")
  if not Hash.verify(user.password, request.password):
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Incorrect password")
  
  access_token = oauth2.create_access_token(data={'sub': user.username})

  return {
    'access_token': access_token,
    'token_type': 'bearer',
    'user_id': user.id,
    'username': user.username
  }
from fastapi.security import OAuth2PasswordBearer
from typing import Optional
from datetime import datetime, timedelta
from jose import jwt
 
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
SECRET_KEY = '77407c7339a6c00544e51af1101c4abb4aea2a31157ca5f7dfd87da02a628107'
ALGORITHM = 'HS256'
ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
  to_encode = data.copy()
  if expires_delta:
    expire = datetime.utcnow() + expires_delta
  else:
    expire = datetime.utcnow() + timedelta(minutes=15)
  to_encode.update({"exp": expire})
  encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  return encoded_jwt

 

   2) token 확인 endpoint 호출 및 JWT 토큰 검증 

@router.get('/', response_model=List[UserDisplay])
def get_all_users(db: Session = Depends(get_db), current_user: UserBase = Depends(get_current_user)):
  return db_user.get_all_users(db)

 

      - OAuth2PasswordBearer의 입력 인자인 tokenUrl은 토큰이 생성되는 엔드포인트를 문서에서 연결하여 알려주기 위한 값으로, 일반적으로 login과 관련된 엔드포인트를 지정한다. 없으면 공란으로 둬도 상관 없다.

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
  credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail='Could not validate credentials',
    headers={"WWW-Authenticate": "Bearer"}
  )
  try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    username: str = payload.get("sub")
    if username is None:
      raise credentials_exception
  except JWTError:
    raise credentials_exception
  
  user = db_user.get_user_by_username(db, username)

  if user is None:
    raise credentials_exception

  return user

 

   4) OAuth2PasswordRequestForm과 OAuth2PasswordBearer 정리

      1. OAuth2PasswordRequestForm

         - 설명 : 

            - OAuth2의 "Password Grant"를 지원하는 입력 데이터 형식
            - 사용자가 username과 password를 전송하면, 서버는 이를 받아 유효성을 검증함

               - 토큰을 반환하지 않는다.
            - FastAPI에서 요청 바디로 제공되며, username, password, scope 등의 필드를 자동으로 처리

 

         - 주요 필드 : 

            - username: 사용자 ID 또는 이메일
            - password: 사용자의 비밀번호
            - scope: 인증 범위를 지정하는 문자열 (선택적)

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

# 임시 사용자 데이터베이스
fake_users_db = {
    "user@example.com": {
        "username": "user@example.com",
        "password": "securepassword",
    }
}

class Token(BaseModel):
    access_token: str
    token_type: str

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(status_code=400, detail="Invalid credentials")
    
    return {"access_token": f"token-for-{form_data.username}", "token_type": "bearer"}

 

      2. OAuth2PasswordBearer

         - 설명 :
            - OAuth2 Bearer Token 인증을 처리하는 보안 스키마
            - 클라이언트가 Authorization 헤더에 Bearer Token을 포함하여 요청을 보낼 때 사용
            - 토큰이 없거나 유효하지 않으면 자동으로 인증 실패 응답을 반환

         - 주요 역할 :
            - API 엔드포인트에서 인증이 필요한 경우 토큰을 검사

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

# OAuth2PasswordBearer 초기화 (토큰 발급 경로를 지정)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 임시 사용자 데이터베이스
fake_users_db = {
    "user@example.com": {"username": "user@example.com", "password": "securepassword"}
}

# 토큰 모델
class Token(BaseModel):
    access_token: str
    token_type: str

# 토큰 발급 엔드포인트
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(status_code=400, detail="Invalid username or password")
    
    # 성공적으로 인증된 경우 토큰 발급
    return {"access_token": "valid-token", "token_type": "bearer"}

# 토큰 검증 함수
async def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"username": "user@example.com"}

# 보호된 엔드포인트
@app.get("/secure-data")
async def secure_data(user: dict = Depends(get_current_user)):
    return {"message": f"Hello {user['username']}, this is secure data!"}

 

   5) Scope & Scopes

      - Scopes는 OAuth2에서 특정 리소스 또는 기능에 대한 접근 권한을 정의하는 데 사용된다.

         - 예: read, write, admin 같은 권한 레벨을 지정

 

      - 특징:
         - 사용자의 역할이나 권한을 세분화하여 관리
         - 클라이언트가 특정 스코프를 요청하고, 서버는 이를 제한적으로 허용
         - FastAPI에서 scopes는 토큰 발급 시 포함되며, 이후 요청에서 검증된다.

 

      - 사용 예제: Scopes

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from typing import List

# OAuth2PasswordBearer를 초기화하며 scopes 정의
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", scopes={"read": "Read access", "write": "Write access"})

app = FastAPI()

# Mock 데이터베이스
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "hashed_password": "fakehashedpassword",
        "scopes": ["read"],  # johndoe는 "read" 권한만 가짐
    }
}

# 사용자 인증 함수
def authenticate_user(username: str, password: str):
    user = fake_users_db.get(username)
    if not user or user["hashed_password"] != "fakehashedpassword":
        return None
    return user

# 토큰 검증 함수
def get_current_user(token: str = Depends(oauth2_scheme)):
    # 실제 구현에서는 토큰 해석 및 검증을 수행
    if token == "fake-token-read":
        return {"username": "johndoe", "scopes": ["read"]}
    elif token == "fake-token-write":
        return {"username": "janedoe", "scopes": ["write"]}
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
        )

# 특정 스코프 검증
def get_current_active_user(required_scopes: List[str], user: dict = Depends(get_current_user)):
    for scope in required_scopes:
        if scope not in user["scopes"]:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing scope: {scope}",
            )
    return user

# 토큰 발급 엔드포인트
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Invalid username or password")
    return {"access_token": "fake-token-read" if "read" in user["scopes"] else "fake-token-write", "token_type": "bearer"}

# "read" 스코프가 필요한 엔드포인트
@app.get("/items/")
async def read_items(user: dict = Depends(lambda: get_current_active_user(["read"]))):
    return {"message": "You have 'read' access"}

# "write" 스코프가 필요한 엔드포인트
@app.post("/items/")
async def write_items(user: dict = Depends(lambda: get_current_active_user(["write"]))):
    return {"message": "You have 'write' access"}

 

      - Scope와 Scopes에 대한 정리

      1. Scope
         - 정의 :
            - Scope는 단일 권한 또는 접근 범위를 나타낸다.

from fastapi import FastAPI, Depends, HTTPException

# 가상의 사용자 토큰 데이터베이스
fake_db_tokens = {
    "user1_token": {"username": "user1", "scope": "read"},
    "user2_token": {"username": "user2", "scope": "write"},
}

app = FastAPI()

# 현재 사용자 가져오기
def get_current_user(token: str):
    user = fake_db_tokens.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# 단일 Scope 검증 유틸리티
def verify_scope(required_scope: str):
    def dependency(user: dict = Depends(get_current_user)):
        if user["scope"] != required_scope:
            raise HTTPException(status_code=403, detail=f"Forbidden: {required_scope} scope required")
        return user
    return dependency

# 읽기 권한이 필요한 엔드포인트
@app.get("/read-data/")
def read_data(
    user: dict = Depends(get_current_user),
    verified_user: dict = Depends(verify_scope("read"))
):
    return {"message": f"User {user['username']} can read data"}

# 쓰기 권한이 필요한 엔드포인트
@app.post("/write-data/")
def write_data(
    user: dict = Depends(get_current_user),
    verified_user: dict = Depends(verify_scope("write"))
):
    return {"message": f"User {user['username']} can write data"}

 

      2. Scopes
         - 정의 :
            - 여러 권한(Scope)을 묶은 권한 세트이다.

from fastapi import FastAPI, Depends, HTTPException

# 가상의 사용자 토큰 데이터베이스
fake_db_tokens = {
    "user1_token": {"username": "user1", "scopes": ["read:users", "write:posts"]},
    "user2_token": {"username": "user2", "scopes": ["admin"]},
}

app = FastAPI()

# 현재 사용자 가져오기
def get_current_user(token: str):
    user = fake_db_tokens.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# 복수 Scopes 검증 유틸리티
def verify_scopes(required_scopes: list):
    def dependency(user: dict = Depends(get_current_user)):
        user_scopes = user["scopes"]
        if not any(scope in user_scopes for scope in required_scopes):
            raise HTTPException(status_code=403, detail=f"Forbidden: One of {required_scopes} scopes required")
        return user
    return dependency

# 사용자 읽기 엔드포인트 (read:users 권한 필요)
@app.get("/users/")
def read_users(
    user: dict = Depends(get_current_user),
    verified_user: dict = Depends(verify_scopes(["read:users", "admin"]))
):
    return {"message": f"User {user['username']} can read users"}

# 게시물 작성 엔드포인트 (write:posts 권한 필요)
@app.post("/posts/")
def create_post(
    user: dict = Depends(get_current_user),
    verified_user: dict = Depends(verify_scopes(["write:posts", "admin"]))
):
    return {"message": f"User {user['username']} can create posts"}

 

       3. SecurityScopes란
         - 정의 :

            - 주로 OAuth2PasswordBearer나 OAuth2PasswordRequestForm과 함께 사용되며, 사용자가 가진 스코프를 확인하고 특정 엔드포인트에서 요구하는 스코프와 비교하여 권한을 검증한다.

 

         - 주요 기능 :

            - 복수 스코프 검증: 엔드포인트가 여러 스코프를 요구하는 경우 이를 검증한다.
            - 필요한 스코프 명시: required_scopes로 엔드포인트에서 필요한 스코프를 정의할 수 있다.
            - 사용자 스코프와 비교: 사용자의 스코프와 엔드포인트의 요구 스코프를 비교하여 인증 및 권한 여부를 결정한다.

 

         - 복합적인 스코프 예시

            - read:users → 사용자 정보를 읽을 권한
            - write:posts → 게시물을 작성할 권한
            - admin → 관리자 권한
            - read:posts → 게시물을 읽을 권한

 

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes

# 가상의 사용자 토큰 데이터베이스
fake_db_tokens = {
    "user1_token": {"username": "user1", "scopes": ["read:users", "write:posts"]},
    "user2_token": {"username": "user2", "scopes": ["admin"]},
    "user3_token": {"username": "user3", "scopes": ["read:users", "read:posts"]},
}

# OAuth2 설정
oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={
        "read:users": "Read user information",
        "write:posts": "Write new posts",
        "admin": "Administrative privileges",
        "read:posts": "Read posts"
    }
)

app = FastAPI()

# 현재 사용자 가져오기
def get_current_user(
    security_scopes: SecurityScopes,  # SecurityScopes로 필요한 스코프 가져옴
    token: str = Depends(oauth2_scheme)
):
    user = fake_db_tokens.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")

    # 필요한 스코프 검증
    user_scopes = user.get("scopes", [])
    for scope in security_scopes.scopes:
        if scope not in user_scopes:
            raise HTTPException(
                status_code=403,
                detail=f"Forbidden: Missing required scope {scope}",
            )
    return user

# 사용자 읽기 (read:users 및 read:posts 스코프 필요)
@app.get("/users/")
def read_users(
    current_user: dict = Security(get_current_user, scopes=["read:users", "read:posts"])
):
    return {"message": f"User {current_user['username']} can read users and posts"}

# 게시물 작성 (write:posts 및 read:users 스코프 필요)
@app.post("/posts/")
def create_post(
    current_user: dict = Security(get_current_user, scopes=["write:posts", "read:users"])
):
    return {"message": f"User {current_user['username']} can create posts"}

 

8. 라우팅 역추적 (reverse routing), URL 역참조

   1) 템플릿 내에서 url_for() 사용 : 

      - 이 경우에는 url_for()가 자동으로 현재 애플리케이션의 url_for()를 사용하여 동작한다.
      - url_for()는 Request 객체와 연결되어 있어, 템플릿에서 FastAPI 애플리케이션의 URL을 동적으로 생성할 수 있다.
      - 템플릿에서 url_for()를 사용하여 동적으로 URL을 생성할 때는 Request 객체가 자동으로 처리된다.

      - 템플릿에서 url_for()만 사용해도 동작한다. 템플릿은 Request 객체와 연결되어 있기 때문에 app.url_for()를 사용할 필요가 없다. url_for()는 템플릿에서 자동으로 동작한다.

from fastapi import FastAPI
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from fastapi import Request

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/products/{product_id}", name="view_product")
async def get_product(product_id: int):
    return {"product_id": product_id}

@app.get("/")
async def home(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})
<a href="{{ url_for('view_product', product_id=123) }}">View Product 123</a>

 

   2) 애플리케이션 코드 내에서 app.url_for() 사용 :

      - app.url_for()는 FastAPI 애플리케이션 코드 내에서 특정 라우트의 URL을 동적으로 생성할 때 사용된다.
      - url_for()와 비슷하게 동작하지만, app 객체를 통해 호출해야 한다.
      - 애플리케이션 코드 내에서 app.url_for()를 사용하여 동적으로 URL을 생성하려면 app 객체를 통해 호출해야 한다.

      - 애플리케이션 코드 내에서는 app.url_for()를 사용해야 한다. 이 경우 app 객체를 통해 라우트의 URL을 동적으로 생성한다. url_for()는 템플릿 내에서만 동작하는 기본 방식이고, 코드 내에서는 app.url_for()를 사용해야 한다.

from fastapi import FastAPI

app = FastAPI()

@app.get("/products/{product_id}", name="view_product")
async def get_product(product_id: int):
    return {"product_id": product_id}

@app.get("/get_product_url")
async def get_product_url():
    # 'view_product' 엔드포인트의 URL을 동적으로 생성
    product_url = app.url_for("view_product", product_id=123)
    return {"product_url": product_url}

 

   3) 기본적인 차이점 :

      - 템플릿 내에서 url_for() : 

         - 템플릿에서 url_for()는 Jinja2 템플릿 엔진에서 바로 사용되며, 템플릿 렌더링 시 Request 객체에 자동으로 바인딩된다. 이 경우 app.url_for()를 사용할 필요 없이 url_for()만 호출하면 된다.

      - 애플리케이션 코드 내에서 app.url_for() :

         - FastAPI 애플리케이션 코드 내에서 엔드포인트의 URL을 동적으로 생성하려면 app.url_for()를 사용해야 한다. 이는 특정 라우트 함수의 URL을 코드에서 직접 생성할 때 사용된다.

댓글