FastAPI-Users with MongoDB: JWT 인증

# 설치
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를 사용할 때의 주요 장점

1. 스키마 유연성
   • 동적 스키마 지원
   • 복잡한 데이터 구조 저장 가능
2. 확장성
   • 수평적 확장 용이
   • 대량의 사용자 데이터 처리
3. 쿼리 성능
   • 인덱싱 지원
   • 빠른 읽기/쓰기 작업
4. 통합 용이성
   • Beanie ORM 사용으로 쉬운 통합
   • 비동기 작업 지원

FastAPI-Users와 MongoDB를 함께 사용하면 확장 가능하고 유연한 사용자 인증 시스템을 구축할 수 있습니다.