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) — 데이터 스코프 키 id: str # users.id (로그인 아이디) — get_me 재조회 키 company_id: str # users.company_id (uuid) — 멀티테넌트 스코프 키 role: int # users.role (UserRole) — 권한 게이트(최고관리자 등) 판단 키 def __init__(self, *args, **kwargs) -> None: super().__init__() # 구버전 토큰(role 미포함) 도 디코딩되도록 기본값을 먼저 깔고 kwargs 로 덮어쓴다. self.role = UserRole.USER.value 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__)