60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
from fastapi import HTTPException, Request
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from pydantic import BaseModel
|
|
from sqlmodel import SQLModel
|
|
from starlette.status import HTTP_403_FORBIDDEN, HTTP_401_UNAUTHORIZED
|
|
from common.core.config import settings
|
|
from fastapi.security.utils import get_authorization_scheme_param
|
|
from typing import Generic, TypeVar, Optional
|
|
class TokenPayload(BaseModel):
|
|
account: str | None = None
|
|
id: int | None = None
|
|
oid: int | None = None
|
|
|
|
class Token(SQLModel):
|
|
access_token: str
|
|
token_type: str = "bearer"
|
|
|
|
class XOAuth2PasswordBearer(OAuth2PasswordBearer):
|
|
async def __call__(self, request: Request) -> Optional[str]:
|
|
authorization = request.headers.get(settings.TOKEN_KEY)
|
|
if request.headers.get(settings.ASSISTANT_TOKEN_KEY):
|
|
authorization = request.headers.get(settings.ASSISTANT_TOKEN_KEY)
|
|
scheme, param = get_authorization_scheme_param(authorization)
|
|
|
|
if not authorization or scheme.lower() not in ["bearer", "assistant"]:
|
|
if self.auto_error:
|
|
raise HTTPException(
|
|
status_code=HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
else:
|
|
return None
|
|
return param
|
|
|
|
|
|
|
|
|
|
T = TypeVar('T')
|
|
|
|
class PaginationParams(BaseModel):
|
|
page: int = 1
|
|
size: int = 20
|
|
order_by: Optional[str] = None
|
|
desc: bool = False
|
|
|
|
class PaginatedResponse(BaseModel, Generic[T]):
|
|
items: list[T]
|
|
total: int
|
|
page: int
|
|
size: int
|
|
total_pages: int
|
|
|
|
|
|
class BaseCreatorDTO(BaseModel):
|
|
id: int
|
|
class Config:
|
|
json_encoders = {
|
|
int: lambda v: str(v) if isinstance(v, int) and v > (2**53 - 1) else v
|
|
} |