65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
# -*- 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)
|