Steven
0c53aa7062
- backend: FastAPI 应用工厂、统一错误信封、JWT、短信登录、工作区/成员/邀请、幂等中间件 - alembic 0001: workspace/member/invitation/audit_log/app_user + RLS 策略 + 双角色 GRANT - tests: 认证/工作区/租户隔离/幂等 四组用例（尚未在本机执行） - web: React18+Vite 骨架（登录、工作区、邀请落地页） - android: Compose 骨架（登录、工作区列表） - 2.Design: PRODUCT-PLAN / ROADMAP-M0-M1 / NAMING / DEVELOPMENT
78 lines
2.0 KiB
Python
78 lines
2.0 KiB
Python
"""统一响应体 {code, message, data, request_id} + 游标分页。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Any, Generic, TypeVar
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from .errors import AppError, ErrorCode
|
|
from .logging_setup import request_id_ctx
|
|
|
|
T = TypeVar("T")
|
|
|
|
DEFAULT_LIMIT = 20
|
|
MAX_LIMIT = 100
|
|
|
|
|
|
class Envelope(BaseModel, Generic[T]):
|
|
code: int = 0
|
|
message: str = "ok"
|
|
data: T | None = None
|
|
request_id: str = "-"
|
|
|
|
|
|
class Page(BaseModel, Generic[T]):
|
|
items: list[T]
|
|
next_cursor: str | None = None
|
|
has_more: bool = False
|
|
|
|
|
|
class Cursor(BaseModel):
|
|
"""(created_at, id) 复合游标：单用 created_at 会在同毫秒记录上漏数据。"""
|
|
|
|
created_at: datetime
|
|
id: UUID
|
|
|
|
|
|
def ok(data: Any = None, message: str = "ok") -> dict[str, Any]:
|
|
return {
|
|
"code": int(ErrorCode.OK),
|
|
"message": message,
|
|
"data": data,
|
|
"request_id": request_id_ctx.get(),
|
|
}
|
|
|
|
|
|
def encode_cursor(created_at: datetime, row_id: UUID) -> str:
|
|
raw = json.dumps({"t": created_at.isoformat(), "i": str(row_id)}, separators=(",", ":"))
|
|
return base64.urlsafe_b64encode(raw.encode()).decode().rstrip("=")
|
|
|
|
|
|
def decode_cursor(cursor: str) -> Cursor:
|
|
try:
|
|
padded = cursor + "=" * (-len(cursor) % 4)
|
|
payload = json.loads(base64.urlsafe_b64decode(padded))
|
|
return Cursor(created_at=datetime.fromisoformat(payload["t"]), id=UUID(payload["i"]))
|
|
except (binascii.Error, KeyError, ValueError, json.JSONDecodeError) as exc:
|
|
raise AppError(ErrorCode.BAD_REQUEST, "分页游标不合法") from exc
|
|
|
|
|
|
def clamp_limit(limit: int) -> int:
|
|
if limit <= 0:
|
|
return DEFAULT_LIMIT
|
|
return min(limit, MAX_LIMIT)
|
|
|
|
|
|
class Pagination(BaseModel):
|
|
limit: int = Field(default=DEFAULT_LIMIT, ge=1, le=MAX_LIMIT)
|
|
cursor: str | None = None
|
|
|
|
def parsed(self) -> tuple[int, Cursor | None]:
|
|
return clamp_limit(self.limit), (decode_cursor(self.cursor) if self.cursor else None)
|