Cheng Zhou
125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
"""权限缓存管理器"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
class PermissionCache:
|
|
"""权限缓存管理器"""
|
|
|
|
def __init__(self, default_ttl: int = 300):
|
|
self.default_ttl = default_ttl
|
|
self._project_permissions_cache: dict[str, tuple[Any, float]] = {}
|
|
self._member_role_cache: dict[str, tuple[str | None, float]] = {}
|
|
|
|
def _is_expired(self, timestamp: float, ttl: int) -> bool:
|
|
return time.time() - timestamp > ttl
|
|
|
|
def _make_project_cache_key(self, study_id: uuid.UUID) -> str:
|
|
return f"project_permissions:{study_id}"
|
|
|
|
def _make_member_cache_key(self, study_id: uuid.UUID, user_id: uuid.UUID) -> str:
|
|
return f"member_role:{study_id}:{user_id}"
|
|
|
|
async def get_member_role(
|
|
self,
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
ttl: int | None = None,
|
|
) -> str | None:
|
|
"""获取成员角色（带缓存）"""
|
|
from app.crud import member as member_crud
|
|
|
|
if ttl is None:
|
|
ttl = self.default_ttl
|
|
|
|
cache_key = self._make_member_cache_key(study_id, user_id)
|
|
|
|
if cache_key in self._member_role_cache:
|
|
cached_role, timestamp = self._member_role_cache[cache_key]
|
|
if not self._is_expired(timestamp, ttl):
|
|
return cached_role
|
|
|
|
membership = await member_crud.get_member(db, study_id, user_id)
|
|
role = membership.role_in_study if membership and membership.is_active else None
|
|
|
|
self._member_role_cache[cache_key] = (role, time.time())
|
|
return role
|
|
|
|
def get_project_permissions(
|
|
self,
|
|
study_id: uuid.UUID,
|
|
ttl: int | None = None,
|
|
) -> dict[str, dict[str, bool]] | None:
|
|
"""获取项目接口权限覆盖表缓存"""
|
|
if ttl is None:
|
|
ttl = self.default_ttl
|
|
|
|
cache_key = self._make_project_cache_key(study_id)
|
|
if cache_key not in self._project_permissions_cache:
|
|
return None
|
|
|
|
permissions, timestamp = self._project_permissions_cache[cache_key]
|
|
if self._is_expired(timestamp, ttl):
|
|
self._project_permissions_cache.pop(cache_key, None)
|
|
return None
|
|
|
|
return permissions
|
|
|
|
def set_project_permissions(
|
|
self,
|
|
study_id: uuid.UUID,
|
|
permissions: dict[str, dict[str, bool]],
|
|
) -> None:
|
|
"""缓存项目接口权限覆盖表"""
|
|
cache_key = self._make_project_cache_key(study_id)
|
|
self._project_permissions_cache[cache_key] = (permissions, time.time())
|
|
|
|
def invalidate_project_permissions(self, study_id: uuid.UUID) -> None:
|
|
cache_key = self._make_project_cache_key(study_id)
|
|
self._project_permissions_cache.pop(cache_key, None)
|
|
|
|
def invalidate_member_role(self, study_id: uuid.UUID, user_id: uuid.UUID) -> None:
|
|
cache_key = self._make_member_cache_key(study_id, user_id)
|
|
self._member_role_cache.pop(cache_key, None)
|
|
|
|
def invalidate_all_member_roles(self, study_id: uuid.UUID) -> None:
|
|
keys_to_delete = [
|
|
key for key in self._member_role_cache.keys()
|
|
if key.startswith(f"member_role:{study_id}:")
|
|
]
|
|
for key in keys_to_delete:
|
|
del self._member_role_cache[key]
|
|
|
|
def clear_all(self) -> None:
|
|
self._project_permissions_cache.clear()
|
|
self._member_role_cache.clear()
|
|
|
|
def get_cache_stats(self) -> dict[str, Any]:
|
|
return {
|
|
"project_permissions_count": len(self._project_permissions_cache),
|
|
"member_role_count": len(self._member_role_cache),
|
|
"total_count": len(self._project_permissions_cache) + len(self._member_role_cache),
|
|
}
|
|
|
|
|
|
_permission_cache: PermissionCache | None = None
|
|
|
|
|
|
def get_permission_cache() -> PermissionCache:
|
|
global _permission_cache
|
|
if _permission_cache is None:
|
|
_permission_cache = PermissionCache()
|
|
return _permission_cache
|
|
|
|
|
|
def set_permission_cache(cache: PermissionCache) -> None:
|
|
global _permission_cache
|
|
_permission_cache = cache
|