OCI Policy Analysis Common Components

Data Models

Legacy re-exports for model modules (compatibility shim).

Display Helper Functions

Legacy re-exports for display helpers (compatibility shim).

oci_policy_analysis.common.helpers.for_display_admit(statement) → dict

Display-friendly dict for parsed AdmitStatement, mapped to UI columns.

oci_policy_analysis.common.helpers.for_display_define(define: DefineStatement) → dict

Return a dictionary suitable for display purposes for defines.

oci_policy_analysis.common.helpers.for_display_dynamic_group(dg: DynamicGroup) → dict

Return a dictionary suitable for display purposes for dynamic groups.

oci_policy_analysis.common.helpers.for_display_endorse(statement) → dict

Display-friendly dict for parsed EndorseStatement, mapped to UI columns.

oci_policy_analysis.common.helpers.for_display_group(g: Group) → dict

Return a dictionary suitable for display purposes for groups.

oci_policy_analysis.common.helpers.for_display_policy(statement: RegularPolicyStatement) → dict

Return a dictionary suitable for display purposes for policy statements.

oci_policy_analysis.common.helpers.for_display_tag_based_policy_row(statement: RegularPolicyStatement) → dict

Return a compact, tag-focused display dict for a regular statement.

oci_policy_analysis.common.helpers.for_display_user(u: User) → dict

Return a dictionary suitable for display purposes for users.

Cache Manager

class oci_policy_analysis.common.caching.CacheManager(cache_dir: Path | None = None)

Bases: object

Handles saving and loading cached JSON data (IAM + AI). Also supports canonical consolidation session persistence: per-tenancy state files for protected_set + plan history. Each cache is tied to a tenancy name and date. The cache directory is ~/.oci-policy-analysis/cache by default, but can be overridden. Caches have the concept of being “preserved” to avoid automatic deletion during culling.

get_or_create_consolidation_state(tenancy_ocid) → dict

Load the canonical consolidation state file for this tenancy_ocid. If it does not exist, create a blank state (protected_set={}, history=[]). Returns the state dict; saves it if it was new.

save_consolidation_state(tenancy_ocid, state: dict)

Overwrite the canonical consolidation state file for this tenancy_ocid.

get_protected_set(tenancy_ocid)

Return the protected_set from consolidation state. Creates file if needed.

set_protected_set(tenancy_ocid, protected_set: dict)

Update and persist the protected_set in the canonical consolidation state file.

get_history(tenancy_ocid)

Return the list of run/plan history for this tenancy_ocid. If no tenancy_ocid, returns empty list (safe for new sessions).

add_run_record(tenancy_ocid, run_record: dict)

Append a new run/plan record to history and save the canonical file.

update_run_record(tenancy_ocid: str, consolidation_effort_id: str, updates: dict) → bool

Update an existing run record in history by consolidation_effort_id.

Parameters:

• tenancy_ocid – Tenancy OCID.

• consolidation_effort_id – Run identifier (plan_id).

• updates – Dict of fields to merge into the run record (e.g. status, step_status).

remove_run_record(tenancy_ocid: str, consolidation_effort_id: str) → bool

Remove a run record from history by consolidation_effort_id.

Parameters:

• tenancy_ocid – Tenancy OCID.

• consolidation_effort_id – Run identifier (plan_id) to remove.

list_consolidation_tenancy_ocids() → list[str]

Return list of tenancy OCIDs that have consolidation state files.

list_all_consolidation_plans() → list[dict]

Return a flat list of all consolidation plans across tenancies. Each item has tenancy_ocid, consolidation_effort_id, created_at, and run (full record).

load_prospects(tenancy_ocid: str) → list[dict[str, Any]]

Load standalone prospective statements for a tenancy from cache/prospects.

save_prospects(tenancy_ocid: str, rows: list[dict[str, Any]]) → None

Persist standalone prospective statements for a tenancy to cache/prospects.

save_consolidation_session(session_data: dict, plan_id: str | None = None, tenancy_ocid: str | None = None, preserved: bool = False) → str

Save a consolidation session (see models_consolidation.ConsolidationSession) to disk as JSON. Returns file name/path. :param session_data: The dict/TypedDict structure matching ConsolidationSession (pluggable, robust to version drift) :param plan_id: Optional label/ID for file naming, else uses timestamp. :param tenancy_ocid: Optional for nested cache structure. :param preserved: Mark this session as not-to-be-deleted.

Returns:

The output file path string.

load_consolidation_session(plan_id: str, tenancy_ocid: str | None = None) → dict | None

Load a consolidation session from cache. :param plan_id: Label/filename (w/o json) for the session/plan. :param tenancy_ocid: Optional subdir

Returns:

session_data (dict) or None

save_combined_cache(policy_analysis: PolicyAnalysisRepository, export_file=None, preserved: bool = False) → str

Save combined cache for policies and dynamic groups. Returns file name.

Parameters:

• export_file – Optional file handle to export to instead of saving to cache directory

• preserved – Whether to mark this cache as preserved (not auto-deleted)

Returns:

The name of the file saved

load_combined_cache(policy_analysis: PolicyAnalysisRepository, named_cache: str) → str

Load combined cache for policies and dynamic groups.

Given the name and data of a cache, loads the data into both of the centralized structures for Compartment/Policy JSON storage.

Parameters:

named_cache – The tenancy_date string of the cache name to load

Returns:

A string indicating the name of the file used

load_cache_from_json(policy_analysis: PolicyAnalysisRepository, loaded_json: dict) → bool

Load combined cache data from a given JSON dict. Given loaded JSON data, loads the data into both of the centralized structures for Compartment/Policy JSON storage.

Parameters:

loaded_json – The loaded JSON data as a dict

get_preserved_cache_set() → set

Get a set of cache names which are marked as preserved.

get_available_cache(tenancy_name: str | None) → list[str]

Get available cache files for a given profile

If given no argument, simply return the list of all avialable cache files that exist in the cache directory. Entries will contain the tenancy name and date loaded.

Parameters:

• tenancy_name – The name of an OCI tenancy, which will filter the cache list down to only

• tenancy. (caches for that)

Returns:

a list of the available named caches

load_cache_into_local_json(cached_tenancy: str) → dict

Takes a named cache (tenancy_date) and returns the loaded JSON data as a dict. Used for exporting or other purposes.

Parameters:

cached_tenancy – The tenancy_date string of the cache name to load

Returns:

The loaded cache data as a dict

remove_cache_entry(named_cache: str) → bool

Remove specified cache file AND its entry from cache_entries.json.

Parameters:

named_cache – The tenancy_date string of the cache name to remove

Returns:

True if both file and entry were removed, False otherwise

rename_cache_entry(old_named_cache: str, new_named_cache: str) → bool

Rename both the cache file and its entry in cache_entries.json.

Parameters:

• old_named_cache – The current tenancy_date string of the cache name

• new_named_cache – The new tenancy_date string of the cache name

Returns:

True if both file and entry were renamed, False otherwise. Returns False (and does NOT rename) if a cache file or entry already exists with the new name.

preserve_cache_entry(named_cache: str, preserve: bool = True) → bool

Mark or unmark a cache entry as preserved in cache_entries.json.

Parameters:

• named_cache – The tenancy_date string of the cache name to update

• preserve – True to mark as preserved, False to unmark

Returns:

True if the entry was updated, False otherwise

update_policy_section(policy_analysis: PolicyAnalysisRepository, policy_data_reloaded: str | None)

Update ONLY the policies, policy_statements, tag catalog, compartments, defined_aliases, and cross_tenancy_statements in the most recent cache file for a given tenancy, and set ‘policy_data_reloaded’ with the supplied timestamp. This preserves IAM/user/group data and other session metadata. No effect if no cache is present.

Parameters:

• policy_analysis – PolicyAnalysisRepository with fresh policy/compartment data in memory

• policy_data_reloaded – ISO timestamp string for reloaded policy data

Global Logger

class oci_policy_analysis.common.logger.ForceFlushStreamHandler(stream=None)

Bases: StreamHandler

A stream handler that flushes after every emit, ensuring logs always show up immediately—even from threads.

emit(record)

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

oci_policy_analysis.common.logger.get_logger(component: str | None = None) → Logger

Return a component logger. No handlers added (propagate to root).

Parameters:

component – Component name (e.g., ‘cli’, ‘data_repo’). If None, uses base name.

Returns:

Logger instance.

oci_policy_analysis.common.logger.set_log_level(level: str | int) → None

Set root logger level (affects all non-overridden loggers).

• Use for global (root) logging threshold, e.g., on app startup or when user changes global log level.

• When –verbose is active, will force DEBUG everywhere (but UI always filters DEBUG).

• Do not use to configure UI/ConsoleTab handler; that always filters at INFO+.

Options:

• CRITICAL 50

• ERROR 40

• WARNING 30

• INFO 20

• DEBUG 10

Parameters:

level – Level name (e.g., ‘DEBUG’) or int.

oci_policy_analysis.common.logger.set_component_level(component: str, level: str | int) → None

Set level for a specific logger (app or third-party). Component can be full logger name (with dots) or just base name. Example: set_component_level(‘cli’, ‘DEBUG’)

Parameters:

• component – Component/logger name (e.g., ‘cli’, ‘data_repo’, ‘requests’)

• level – Level name (e.g., ‘DEBUG’) or int.