diff --git a/backend/common/core/schemas.py b/backend/common/core/schemas.py new file mode 100644 index 0000000..6d9d75f --- /dev/null +++ b/backend/common/core/schemas.py @@ -0,0 +1,60 @@ +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 + } \ No newline at end of file