| 12345678910111213141516171819202122232425262728293031323334353637383940 |
- from __future__ import annotations
- from dataclasses import dataclass
- @dataclass(frozen=True)
- class CacheDependency:
- cache_key_hash: str
- point_key: str
- business_domain_uid: str | None
- generation: int
- @dataclass(frozen=True)
- class CacheInvalidationPlan:
- cache_key_hashes: frozenset[str]
- business_domains_to_flush: frozenset[str]
- def plan_cache_invalidation(
- dependencies: list[CacheDependency],
- *,
- changed_point_keys: set[str],
- impacted_point_keys: set[str],
- dependency_metadata_complete: bool = True,
- affected_business_domains: set[str] | None = None,
- ) -> CacheInvalidationPlan:
- affected_points = changed_point_keys | impacted_point_keys
- cache_keys = frozenset(
- dependency.cache_key_hash
- for dependency in dependencies
- if dependency.point_key in affected_points
- )
- domains = frozenset()
- if not dependency_metadata_complete:
- domains = frozenset(affected_business_domains or ())
- return CacheInvalidationPlan(
- cache_key_hashes=cache_keys,
- business_domains_to_flush=domains,
- )
|
