# -*- coding: utf-8 -*- """공통 FastAPI 의존성: 현재 사용자(필수/선택) + 페이지네이션.""" from dataclasses import dataclass from fastapi import Depends, HTTPException, Query from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from .database import get_session from .models import User from .security import decode_token _bearer = HTTPBearer(auto_error=False) async def _user_from_token(creds: HTTPAuthorizationCredentials | None, session: AsyncSession) -> User | None: if creds is None: return None payload = decode_token(creds.credentials) if not payload or payload.get("type") != "access": return None user = ( await session.execute( select(User).where(User.user_uuid == payload.get("sub"), User.is_deleted == False) # noqa: E712 ) ).scalar_one_or_none() if user is None or not user.is_active: return None return user async def get_current_user( creds: HTTPAuthorizationCredentials | None = Depends(_bearer), session: AsyncSession = Depends(get_session), ) -> User: user = await _user_from_token(creds, session) if user is None: raise HTTPException(401, "로그인이 필요합니다.") return user async def get_current_user_optional( creds: HTTPAuthorizationCredentials | None = Depends(_bearer), session: AsyncSession = Depends(get_session), ) -> User | None: return await _user_from_token(creds, session) @dataclass class Pagination: page: int page_size: int @property def offset(self) -> int: return (self.page - 1) * self.page_size def get_pagination( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), ) -> Pagination: return Pagination(page=page, page_size=page_size)