[udemy] Complete FastAPI masterclass from scratch - 학습 정리 1
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을 코드에서 직접 생성할 때 사용된다.
'Study > fastapi' 카테고리의 다른 글
[udemy] Complete FastAPI masterclass from scratch 학습 평가 (0) | 2025.01.30 |
---|---|
[udemy] Complete FastAPI masterclass from scratch - 학습 정리 2 (0) | 2025.01.30 |
FastAPI에서 CSRF(Cross-Site Request Forgery) 적용하는 방법 (0) | 2025.01.15 |
fastapi + sqlalchemy + ORM을 이용한 페이지네이션 방법들 (0) | 2025.01.14 |
FastAPI의 예외 처리, blinker, SQLAlchemy의 listens_for 정리 (0) | 2025.01.14 |
댓글