65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
"""Projection helpers that do not make user-engine a token issuer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from user_engine.domain import Actor, ProjectionType
|
|
from user_engine.service import Projection, UserEngineService
|
|
|
|
|
|
@dataclass
|
|
class CacheStatus:
|
|
entries: int
|
|
tenants: tuple[str, ...]
|
|
applications: tuple[str, ...]
|
|
users: tuple[str, ...]
|
|
|
|
|
|
@dataclass
|
|
class ClaimsEnrichmentProjectionCache:
|
|
"""Cache claims-enrichment projections for external token adapters.
|
|
|
|
The adapter caches profile material only. A caller that issues tokens must
|
|
still own token minting, signing, lifetimes, and issuer-specific policy.
|
|
"""
|
|
|
|
_cache: dict[tuple[str, str, str, str], Projection] = field(default_factory=dict)
|
|
|
|
def get(
|
|
self,
|
|
service: UserEngineService,
|
|
actor: Actor,
|
|
*,
|
|
user_id: str,
|
|
tenant: str,
|
|
application_id: str,
|
|
correlation_id: str,
|
|
) -> Projection:
|
|
key = (tenant, application_id, user_id, actor.subject)
|
|
cached = self._cache.get(key)
|
|
if cached is not None:
|
|
return cached
|
|
projection = service.projection(
|
|
actor,
|
|
user_id,
|
|
ProjectionType.CLAIMS_ENRICHMENT,
|
|
tenant=tenant,
|
|
application_id=application_id,
|
|
correlation_id=correlation_id,
|
|
)
|
|
self._cache[key] = projection
|
|
return projection
|
|
|
|
def invalidate_user(self, user_id: str) -> None:
|
|
for key in tuple(self._cache):
|
|
if key[2] == user_id:
|
|
del self._cache[key]
|
|
|
|
def status(self) -> CacheStatus:
|
|
return CacheStatus(
|
|
entries=len(self._cache),
|
|
tenants=tuple(sorted({key[0] for key in self._cache})),
|
|
applications=tuple(sorted({key[1] for key in self._cache})),
|
|
users=tuple(sorted({key[2] for key in self._cache})),
|
|
)
|