"""Service facade for cache persistence operations."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from oci_policy_analysis.application.core.support.caching import CacheManager
from oci_policy_analysis.application.core.support.logger import get_logger
[docs]
@dataclass
class CacheService:
"""Expose cache operations to UI/API without direct CacheManager usage."""
cache: CacheManager
def __post_init__(self) -> None:
"""Initialize logger after dataclass construction.
Returns:
None
"""
self.logger = get_logger(component='cache_service')
[docs]
def list_caches(self, tenancy_name: str | None = None) -> list[str]:
"""List available cache names.
Args:
tenancy_name: Optional tenancy name filter.
Returns:
list[str]: Cache names matching the filter.
"""
self.logger.info('Listing caches: tenancy_name=%s', tenancy_name or '')
return self.cache.get_available_cache(tenancy_name)
[docs]
def get_preserved_cache_set(self) -> set[str]:
"""Fetch the set of preserved cache names.
Returns:
set[str]: Preserved cache names.
"""
return self.cache.get_preserved_cache_set()
[docs]
def maintenance_records(self) -> list[dict[str, Any]]:
"""List cache names and their current lock state without reading snapshot data.
Returns:
list[dict[str, Any]]: Names and preserved/locked flags from the cache index.
"""
locked = self.get_preserved_cache_set()
return [{'name': name, 'locked': name in locked} for name in self.list_caches()]
[docs]
def consolidation_records(self) -> list[dict[str, Any]]:
"""List saved consolidation tenancies and counts without loading inventory.
Returns:
list[dict[str, Any]]: Tenancy IDs with plan and protection counts.
"""
rows = []
for tenancy in sorted(self.cache.list_consolidation_tenancy_ocids()):
try:
state = self.cache.read_consolidation_state(tenancy)
except (OSError, ValueError) as exc:
rows.append({'tenancy_ocid': tenancy, 'plans': 'Unknown', 'protected': 'Unknown', 'error': str(exc)})
continue
rows.append(
{
'tenancy_ocid': tenancy,
'plans': len(state.get('history') or []),
'protected': len((state.get('protected_set') or {}).get('protected') or []),
}
)
return rows
[docs]
def reset_consolidation(self, tenancy_ocid: str) -> dict[str, Any]:
"""Clear one tenancy's local protection and complete consolidation history.
Args:
tenancy_ocid: Explicitly selected tenancy; must be nonempty.
Returns:
dict[str, Any]: Tenancy ID and number of removed history records.
Raises:
ValueError: If no tenancy is supplied.
Notes:
Uses the existing canonical state format. Inventory caches and OCI
policies are unchanged. Removes saved notes, progress, and rollback
snapshots along with their plan records; no migration is required.
"""
if not tenancy_ocid or not tenancy_ocid.strip():
raise ValueError('Select a tenancy to reset consolidation data.')
previous = self.cache.get_or_create_consolidation_state(tenancy_ocid)
history_count = len(previous.get('history') or [])
self.cache.save_consolidation_state(tenancy_ocid, {'protected_set': {}, 'history': []})
return {'tenancy_ocid': tenancy_ocid, 'cleared_history_records': history_count}
[docs]
def set_locked(self, cache_name: str, locked: bool) -> bool:
"""Preserve or release a saved cache for automatic and explicit cleanup.
Args:
cache_name: Existing saved cache name.
locked: True to protect the cache, False to allow deletion.
Returns:
bool: Whether the cache lock was updated successfully.
"""
return self.cache.preserve_cache_entry(cache_name, preserve=locked)
[docs]
def load_cache(self, policy_repo, cache_name: str) -> str:
"""Load a named cache into the provided policy repository.
Args:
policy_repo: Policy repository receiving loaded data.
cache_name: Cache entry name.
Returns:
str: Underlying cache manager status/result.
"""
self.logger.info('Loading cache entry: %s', cache_name)
return self.cache.load_combined_cache(policy_repo, cache_name)
[docs]
def load_cache_json(self, cache_name: str) -> dict[str, Any]:
"""Load a named cache file as JSON.
Args:
cache_name: Cache entry name.
Returns:
dict[str, Any]: Loaded JSON content.
"""
return self.cache.load_cache_into_local_json(cache_name)
[docs]
def remove_cache(self, cache_name: str) -> bool:
"""Remove a cache entry.
Args:
cache_name: Cache entry name.
Returns:
bool: True when a cache entry was removed.
"""
self.logger.info('Removing cache entry: %s', cache_name)
return bool(self.cache.remove_cache_entry(cache_name))
[docs]
def rename_cache(self, old_cache_name: str, new_cache_name: str) -> bool:
"""Rename an existing cache entry.
Args:
old_cache_name: Existing cache name.
new_cache_name: New cache name.
Returns:
bool: True when rename succeeded.
"""
self.logger.info('Renaming cache entry: %s -> %s', old_cache_name, new_cache_name)
return bool(self.cache.rename_cache_entry(old_cache_name, new_cache_name))
[docs]
def save_cache(self, policy_repo, export_file: Any | None = None, preserved: bool = False) -> str:
"""Save current repository state into cache storage.
Args:
policy_repo: Policy repository source.
export_file: Optional explicit export destination.
preserved: Whether to mark cache as preserved.
Returns:
str: Saved cache entry name or status.
"""
self.logger.info('Saving cache entry: preserved=%s', preserved)
return self.cache.save_combined_cache(policy_repo, export_file=export_file, preserved=preserved)
[docs]
def import_from_json(self, *, loaded_json: dict[str, Any], policy_repo: Any) -> bool:
"""Import repository state from pre-loaded JSON payload.
Args:
loaded_json: Parsed JSON payload.
policy_repo: Policy repository target.
Returns:
bool: True when import succeeds.
"""
self.logger.info('Importing cache data from JSON payload')
return bool(self.cache.load_cache_from_json(loaded_json=loaded_json, policy_analysis=policy_repo))
[docs]
def update_policy_section(
self,
policy_repo: Any,
*,
policy_data_reloaded: str | None = None,
base_cache_name: str | None = None,
) -> str | None:
"""Create a policy-reload cache from an existing cache snapshot.
Args:
policy_repo: Policy repository source.
policy_data_reloaded: Optional policy data indicator.
base_cache_name: Cache that supplied the retained IAM data.
Returns:
New cache path, or ``None`` when no cache could be created.
"""
return self.cache.update_policy_section(
policy_repo,
policy_data_reloaded=policy_data_reloaded,
base_cache_name=base_cache_name,
)
[docs]
def get_consolidation_state(self, tenancy_ocid: str) -> dict:
"""Get or create consolidation state for a tenancy.
Args:
tenancy_ocid: Tenancy OCID key.
Returns:
dict: Consolidation state object.
"""
return self.cache.get_or_create_consolidation_state(tenancy_ocid)
[docs]
def save_consolidation_state(self, tenancy_ocid: str, state: dict) -> None:
"""Persist consolidation state for a tenancy.
Args:
tenancy_ocid: Tenancy OCID key.
state: Consolidation state payload.
Returns:
None
"""
self.logger.info('Saving consolidation state: tenancy_ocid=%s', tenancy_ocid)
self.cache.save_consolidation_state(tenancy_ocid, state)