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
112 lines
3.2 KiB
Python
112 lines
3.2 KiB
Python
"""短生命周期状态存储：验证码、幂等键。
|
|
|
|
dev/test 用进程内实现（不依赖 Redis 也能跑测试），prod 强制走 Redis。
|
|
接口刻意做小，将来换实现不影响调用方。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from typing import Protocol
|
|
|
|
from .config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
class KVStore(Protocol):
|
|
async def set(self, key: str, value: str, ttl_seconds: int) -> None: ...
|
|
|
|
async def get(self, key: str) -> str | None: ...
|
|
|
|
async def delete(self, key: str) -> None: ...
|
|
|
|
async def incr(self, key: str, ttl_seconds: int) -> int: ...
|
|
|
|
|
|
class MemoryStore:
|
|
"""进程内 TTL 字典。仅限单进程 dev/test——多 worker 下不共享，别在生产用。"""
|
|
|
|
def __init__(self) -> None:
|
|
self._data: dict[str, tuple[str, float]] = {}
|
|
|
|
def _purge(self, key: str) -> None:
|
|
item = self._data.get(key)
|
|
if item and item[1] < time.time():
|
|
self._data.pop(key, None)
|
|
|
|
async def set(self, key: str, value: str, ttl_seconds: int) -> None:
|
|
self._data[key] = (value, time.time() + ttl_seconds)
|
|
|
|
async def get(self, key: str) -> str | None:
|
|
self._purge(key)
|
|
item = self._data.get(key)
|
|
return item[0] if item else None
|
|
|
|
async def delete(self, key: str) -> None:
|
|
self._data.pop(key, None)
|
|
|
|
async def incr(self, key: str, ttl_seconds: int) -> int:
|
|
self._purge(key)
|
|
current = int(self._data[key][0]) if key in self._data else 0
|
|
current += 1
|
|
expire_at = self._data[key][1] if key in self._data else time.time() + ttl_seconds
|
|
self._data[key] = (str(current), expire_at)
|
|
return current
|
|
|
|
|
|
class RedisStore:
|
|
def __init__(self, url: str) -> None:
|
|
from redis.asyncio import Redis # 延迟导入：dev 环境没有 redis 包也能起
|
|
|
|
self._client = Redis.from_url(url, decode_responses=True)
|
|
|
|
async def set(self, key: str, value: str, ttl_seconds: int) -> None:
|
|
await self._client.set(key, value, ex=ttl_seconds)
|
|
|
|
async def get(self, key: str) -> str | None:
|
|
return await self._client.get(key)
|
|
|
|
async def delete(self, key: str) -> None:
|
|
await self._client.delete(key)
|
|
|
|
async def incr(self, key: str, ttl_seconds: int) -> int:
|
|
value = await self._client.incr(key)
|
|
if value == 1:
|
|
await self._client.expire(key, ttl_seconds)
|
|
return int(value)
|
|
|
|
|
|
_store: KVStore | None = None
|
|
|
|
|
|
def get_store() -> KVStore:
|
|
global _store
|
|
if _store is None:
|
|
_store = RedisStore(settings.redis_url) if settings.env == "prod" else MemoryStore()
|
|
return _store
|
|
|
|
|
|
def reset_store_for_tests() -> None:
|
|
"""测试夹具用：清掉进程内状态。"""
|
|
global _store
|
|
_store = None
|
|
|
|
|
|
def dumps(value: object) -> str:
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
# ── 业务键约定 ──────────────────────────────────────────
|
|
def sms_code_key(phone: str) -> str:
|
|
return f"sms:login:{phone}"
|
|
|
|
|
|
def sms_throttle_key(phone: str) -> str:
|
|
return f"sms:throttle:{phone}"
|
|
|
|
|
|
def idempotency_key(user_id: str, key: str) -> str:
|
|
return f"idem:{user_id}:{key}"
|