반응형
# 설치
pip install fastapi-users[beanie]
pip install motor
기본 설정과 모델 정의
from typing import Optional
from beanie import Document
from fastapi_users.db import BeanieBaseUser
from pydantic import EmailStr
class User(BeanieBaseUser, Document):
email: EmailStr
hashed_password: str
is_active: bool = True
is_superuser: bool = False
is_verified: bool = False
class Settings:
name = "users" # MongoDB 컬렉션 이름
MongoDB 연결 설정
from beanie import init_beanie
from motor.motor_asyncio import AsyncIOMotorClient
# MongoDB 연결
async def init_db():
client = AsyncIOMotorClient(
"mongodb://localhost:27017"
)
await init_beanie(
database=client.db_name,
document_models=[User]
)
JWT 인증 설정
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import BeanieUserDatabase
# JWT 인증 설정
SECRET = "your-secret-key"
jwt_authentication = JWTAuthentication(
secret=SECRET,
lifetime_seconds=3600,
tokenUrl="/auth/jwt/login",
)
# 사용자 DB 설정
async def get_user_db():
yield BeanieUserDatabase(User)
# FastAPIUsers 인스턴스 생성
fastapi_users = FastAPIUsers[User, int](
get_user_manager,
[jwt_authentication],
)
사용자 관리자 설정
from fastapi_users import BaseUserManager
from fastapi import Depends
class UserManager(BaseUserManager[User, int]):
reset_password_token_secret = SECRET
verification_token_secret = SECRET
async def on_after_register(self, user: User, request=None):
print(f"User {user.email} has registered.")
async def on_after_forgot_password(
self, user: User, token: str, request=None
):
print(f"User {user.email} has forgot their password. Token: {token}")
async def get_user_manager(user_db=Depends(get_user_db)):
yield UserManager(user_db)
애플리케이션 설정
from fastapi import FastAPI, Depends
app = FastAPI()
@app.on_event("startup")
async def startup_event():
await init_db()
# 라우터 설정
app.include_router(
fastapi_users.get_auth_router(jwt_authentication),
prefix="/auth/jwt",
tags=["auth"],
)
app.include_router(
fastapi_users.get_register_router(),
prefix="/auth",
tags=["auth"],
)
app.include_router(
fastapi_users.get_users_router(),
prefix="/users",
tags=["users"],
)
보호된 엔드포인트 구현
from beanie import PydanticObjectId
@app.get("/protected")
async def protected_route(user: User = Depends(fastapi_users.current_user())):
# MongoDB의 ObjectId 사용
return {
"id": str(user.id),
"email": user.email,
"is_active": user.is_active
}
@app.get("/admin")
async def admin_route(
user: User = Depends(fastapi_users.current_user(superuser=True))
):
return {"message": f"Welcome admin {user.email}"}
MongoDB 쿼리 예시
from beanie import PydanticObjectId
@app.get("/users/{user_id}")
async def get_user_by_id(
user_id: PydanticObjectId,
current_user: User = Depends(fastapi_users.current_user(superuser=True))
):
user = await User.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.get("/users/search")
async def search_users(
email: str = None,
is_active: bool = None,
current_user: User = Depends(fastapi_users.current_user(superuser=True))
):
query = {}
if email:
query["email"] = {"$regex": email, "$options": "i"}
if is_active is not None:
query["is_active"] = is_active
users = await User.find(query).to_list()
return users
커스텀 사용자 필드 추가
class User(BeanieBaseUser, Document):
email: EmailStr
hashed_password: str
is_active: bool = True
is_superuser: bool = False
is_verified: bool = False
# 커스텀 필드
full_name: Optional[str] = None
phone_number: Optional[str] = None
created_at: datetime = Field(default_factory=datetime.utcnow)
class Settings:
name = "users"
indexes = [
"email",
"phone_number",
("email", "is_active")
]
MongoDB를 사용할 때의 주요 장점
- 스키마 유연성
- 동적 스키마 지원
- 복잡한 데이터 구조 저장 가능
- 확장성
- 수평적 확장 용이
- 대량의 사용자 데이터 처리
- 쿼리 성능
- 인덱싱 지원
- 빠른 읽기/쓰기 작업
- 통합 용이성
- Beanie ORM 사용으로 쉬운 통합
- 비동기 작업 지원
FastAPI-Users와 MongoDB를 함께 사용하면 확장 가능하고 유연한 사용자 인증 시스템을 구축할 수 있습니다.
'파이썬 > Fast API' 카테고리의 다른 글
SlowAPI: FastAPI에서 Rate Limiting 구현 (0) | 2025.02.23 |
---|---|
FastAPI Permissions: 권한 관리 구현 (0) | 2025.02.22 |
pyinstrument : 성능 최적화를 위한 프로파일링 (0) | 2025.02.11 |
FastAPI 쿼리 매개변수 (0) | 2025.02.09 |
FastAPI와 Nginx : 웹 서버 구성 (1) | 2025.02.07 |