50 lines
1.5 KiB
Python
50 lines
1.5 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 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]
|