import json from typing import Optional from fastapi import Query from pydantic import BaseModel, Field from common.enums import ErrorType, UserRole class StructModel: """프로토콜/구조체 식별용 마커 클래스.""" pass class ErrorInfo(BaseModel, StructModel): """모든 응답에 공통으로 실리는 결과 정보. result.success / code / desc 로 내려간다.""" success: Optional[bool] = True code: Optional[int] = ErrorType.SUCCESS.value desc: Optional[str] = ErrorType.SUCCESS.name def SetResult(self, enum: ErrorType): if enum is not None: self.success = ErrorType.SUCCESS.value == enum.value self.code = enum.value self.desc = enum.name # ---- Protocol 규약 ------------------------------------------------------- # 모든 통신 패킷은 WebPacketProtocol 을 상속한다. # 요청 : Req_xxx (WebPacketProtocol) # 응답 : Res_xxx (Res_WebPacketProtocol) - 항상 result 필드를 가진다. # 각 라우터 폴더의 protocol.py 에 Req_/Res_ 를 정의한다. class WebPacketProtocol(BaseModel, StructModel): pass class Req_WebPacketProtocol(WebPacketProtocol): pass class Res_WebPacketProtocol(WebPacketProtocol): # default_factory 로 인스턴스마다 새 ErrorInfo 를 생성한다 (mutable default 공유 방지). result: ErrorInfo = Field(default_factory=ErrorInfo) msg: Optional[str] = None class Res_PageProtocol(Res_WebPacketProtocol): # 목록 응답 공용 페이지 메타. total: int = 0 page: int = 0 size: int = 0 class PageParams: # 목록 엔드포인트 공용 쿼리 파라미터. 라우터에서 Depends() 로 주입한다. def __init__(self, page: int = Query(1, ge=1), size: int = Query(20, ge=1, le=100)): self.page = page self.size = size @property def skip(self) -> int: return (self.page - 1) * self.size class UserInfo(StructModel): """JWT subject 로 인코딩되는 유저 식별 정보.""" user_id: str # users.user_id (uuid) — 데이터 스코프 키. 사업장은 owner_user_id 로 이 값에 매인다 id: str # users.id (로그인 아이디) — get_me 재조회 키 role: int # users.role (UserRole) — 권한 게이트(최고관리자 등) 판단 키 token_version: int # users.token_version — refresh 토큰 무효화 키(auth_service.refresh_token 이 대조) def __init__(self, *args, **kwargs) -> None: super().__init__() # 구버전 토큰(role·token_version 미포함) 도 디코딩되도록 기본값을 먼저 깔고 kwargs 로 # 덮어쓴다. token_version 기본값은 DB 컬럼 기본값(1)과 같아야 한다 — 배포 순간 옛 # 토큰이 전부 "버전이 다르다"로 거절되는 것을 막는다. self.role = UserRole.USER.value self.token_version = 1 for dictionary in args: for key in dictionary: setattr(self, key, dictionary[key]) for key in kwargs: setattr(self, key, kwargs[key]) def to_json(self) -> str: return json.dumps(self.__dict__)