utils.cache package

Subpackages

• utils.cache.users package
  • Submodules
  • utils.cache.users.placeholders module
    • substitute_primary_domain_placeholder()
    • substitute_scalar_placeholders()
  • Module contents
    • get_merged_users()
    • get_user_defaults()

Submodules

utils.cache.applications module

Application-domain cache: variants, defaults, merged lookup.

Owns _APPLICATIONS_DEFAULTS_CACHE, _VARIANTS_CACHE, and _MERGED_APPLICATIONS_CACHE. Public API: get_application_defaults, get_variants, get_merged_applications. Strictly ansible-free at import time so the GitHub Actions runner-host CLI path (cli.administration.deploy.development.init -> plan_dev_inventory_matrix -> get_variants) keeps working without ansible installed.

utils.cache.applications.get_application_defaults(*, roles_dir: str | os.PathLike[str] | None = None) → dict[str, Any]

utils.cache.applications.get_canonical_volumes(application_id: str) → dict[str, Any]

Return the canonical dict-of-dicts meta/volumes.yml for application_id.

Lives outside the applications payload so its embedded Jinja source: strings stay raw — see the _CANONICAL_VOLUMES_BY_ROLE doc-comment.

An empty result is cached under its id too, or a role without the file would re-stat it on every call.

utils.cache.applications.get_merged_applications(*, variables: dict[str, Any] | None = None, roles_dir: str | os.PathLike[str] | None = None, templar: Any = None) → dict[str, Any]

utils.cache.applications.get_variant_overrides_only(*, roles_dir: str | os.PathLike[str] | None = None) → dict[str, list[dict[str, Any]]]

Return {application_id: [override_0, ...]}, the raw meta/variants.yml entries per role WITHOUT deep-merging meta/services.yml.

Used by matrix-deploy’s –vars bake so the host_vars payload stays sparse: only the variant-specific overrides land there, leaving meta/services.yml fields (notably image/version) blank in host_vars so apply_mirror_overrides can populate them from mirrors.yml.

utils.cache.applications.get_variants(*, roles_dir: str | os.PathLike[str] | None = None) → dict[str, list[Any]]

Return {application_id: [variant_0, ...]} cached per roles_dir. Each variant is the role’s effective configuration after the corresponding meta/variants.yml override has been deep-merged on top of meta/services.yml.

utils.cache.base module

Cross-cutting helpers for the utils.cache.* package.

Holds primitives that the domain modules (users, applications, domains) all need but no domain owns: filesystem constants, YAML loaders, deep-merge, cache-key + content-fingerprint signatures, templar-render machinery, and the cross-domain re-entry guard.

Stays import-cheap so that utils.cache.applications (consumed on GitHub Actions runner hosts that ship without ansible — see CI runs 24934007615 / 24935979190) can pull this module without dragging ansible in. The single ansible-coupled symbol exposed from here is _render_with_templar, which lazy-imports utils.templating.ansible only when actually invoked.

utils.cache.carrier module

Play-scoped carrier of the rendered applications payload.

Ansible forks a fresh worker process per task, so the in-process _MERGED_APPLICATIONS_CACHE starts empty in every worker. The constructor stage renders once and parks the payload plus its cache key in the host fact named by APPLICATIONS_RENDERED_FACT; every later worker inherits that fact at fork time and get_merged_applications serves it instead of rendering. Ansible-free at import time like the rest of the package.

utils.cache.carrier.merged_applications_cache_key(variables: Mapping[str, Any] | None, *, roles_dir: str | os.PathLike[str] | None = None) → tuple[str, tuple]

Key under which one rendered applications payload stays valid.

Parameters:

• variables – The play’s variable scope. Only the subset that _stable_variables_signature reads, and the tokens file it resolves to, influence the key.

• roles_dir – Roles tree the payload is built from; defaults to the repository’s roles/.

Returns:

(resolved roles dir, stable variables signature extended by the tokens file's path and content digest).

utils.cache.carrier.revalidate_tokens_file(variables: Mapping[str, Any] | None) → None

Refresh an NFS client’s attributes of the tokens file before a carrier is keyed.

Parameters:

variables – The play’s variable scope the tokens file resolves from.

utils.cache.domains module

Domain-name cache: canonical-domains map derived from merged apps.

Owns _MERGED_DOMAINS_CACHE. Public API: get_merged_domains. The domain map is intentionally derived from the applications view rather than living in a parallel top-level overrides path. Per-app domain declarations live in applications.<app>.domains and flow through the regular applications-merge pipeline.

utils.cache.domains.get_merged_domains(*, variables: dict[str, Any] | None = None, roles_dir: str | os.PathLike[str] | None = None, templar: Any = None) → dict[str, Any]

Build the canonical-domain map lazily from the merged applications view.

The result is canonical_domains_map(applications, DOMAIN_PRIMARY). Per-app domain declarations live in applications.<app>.domains (canonical/aliases) and flow through the regular applications-merge pipeline.

Cached keyed on (roles_dir, variables_signature, render_state). A read issued while the applications render is in progress sees the unrendered tree, whose service flags are still Jinja strings; its map is cached under its own key so the render’s own call sites reuse it without it ever being served as the finished map. The tor flags are rendered before they are read (_resolved_flag), so that mid-render map carries the same onion domains the finished one does.

utils.cache.files module

Process-level cached filesystem helpers for project-tree scans.

Many integration/lint tests and CLI tools walk the whole repo and read every matching file. Doing that N times per process is wasteful — the tree doesn’t change during a run. These helpers walk the tree once per process and memoise both the path list and file contents, so every caller after the first one pays near-zero cost.

Usage:

from utils.cache.files import iter_project_files, read_text

for path in iter_project_files(extensions=(“.yml”, “.yml.j2”), exclude_tests=True):

content = read_text(path) …

PROJECT_ROOT is the repo root derived from this file’s location.

Gitignore-filtered iteration

The matcher itself lives in utils.cache.gitignore (single SPOT for .gitignore parsing and matching). iter_non_ignored_files() here composes the cached project walk with that matcher and is the recommended helper for lint tests that want the “files git would track” view without subprocessing.

utils.cache.files.iter_non_ignored_files(*, extensions: Iterable[str] | None = None, exclude_tests: bool = False, exclude_dirs: Iterable[str] | None = None, root: str | None = None) → Iterator[str]

Yield project files filtered through .gitignore.

Combines iter_project_files() with is_path_gitignored() so callers get the “files git would track” view without spawning a git subprocess. root defaults to PROJECT_ROOT.

A path is yielded iff it survives both the in-process skip-dir pruning of iter_project_files() and the .gitignore match.

utils.cache.files.iter_project_files(*, extensions: Iterable[str] | None = None, exclude_tests: bool = False, exclude_dirs: Iterable[str] | None = None) → Iterator[str]

Yield absolute paths of project files matching the filter.

Parameters:

• extensions – Suffix strings (e.g. (".yml", ".yaml.j2")). Matching is case-insensitive via plain string suffix. None yields all files.

• exclude_tests – Skip everything under <root>/tests/.

• exclude_dirs – Additional top-level-or-nested directory names to skip. Matched by path-component containment, not by prefix. E.g. exclude_dirs=("docs",) skips any file whose path contains a docs segment.

utils.cache.files.iter_project_files_with_content(*, extensions: Iterable[str] | None = None, exclude_tests: bool = False, exclude_dirs: Iterable[str] | None = None) → Iterator[tuple[str, str]]

Yield (path, content) for matching files; content is cached.

Files that fail to read (permission errors, non-UTF-8) are silently skipped — that matches the prior caller behaviour of wrapping open().read() in a bare try/except Exception: continue.

utils.cache.files.read_text(path: str) → str

Return the UTF-8 text content of a file; cached process-wide.

Raises OSError / UnicodeDecodeError unchanged — callers that scan arbitrary files can wrap this in try/except if non-UTF-8 blobs may appear.

utils.cache.gitignore module

Gitignore-aware path matching for project-tree scans.

SPOT for the fnmatch-based .gitignore-style filter used by lint and integration tests that need to walk the repository without spawning git. Covers the shapes the repo’s .gitignore actually uses (basename patterns, path-qualified patterns, trailing-slash-as-directory) and intentionally omits negation (!) and ** globstar — adding either would require a real pathspec implementation.

Public API:

• load_gitignore_patterns() — read non-empty, non-comment, non-negated lines from <root>/.gitignore. Cached process-wide.

• is_path_gitignored() — pure matcher: does a repo-relative POSIX-style path match any of the supplied patterns?

The combined “yield project files filtered through .gitignore” helper lives in utils.cache.files (iter_non_ignored_files) because it composes the cached project walk with this matcher.

utils.cache.gitignore.is_path_gitignored(rel_path: str, patterns: Iterable[str]) → bool

Return True iff rel_path (POSIX-style, repo-root-relative) is matched by any of patterns.

Supports the shapes the repo’s .gitignore actually uses:

• Bare-basename patterns (no slash): match any path component via fnmatch.

• Path-qualified patterns (with slash): match from the repo root via fnmatch, plus prefix/* shorthand and bare-prefix directory matching.

• Trailing / (directory-only): treated as the stripped pattern because we only test against file paths.

Behaviour intentionally diverges from real git pathspec: * here matches across / (fnmatch semantics). If you need full fidelity, use a dedicated pathspec library.

utils.cache.gitignore.load_gitignore_patterns(root: str | None = None) → tuple[str, ...]

Load non-comment, non-blank, non-negated patterns from <root>/.gitignore.

Returns a tuple (immutable, hashable for lru_cache). Pass root as a string so the cache key is plainly hashable; defaults to PROJECT_ROOT when omitted.

Negation (!) and ** globstar are intentionally not supported — the repo’s .gitignore does not use them, and supporting either would require a real pathspec implementation. If a future .gitignore adds them, extend this helper rather than duplicating the logic in callers.

utils.cache.schema_credentials module

Extract literal default: credential values from a meta/secrets.yml credentials: tree.

Kept in its own module so utils.cache.applications stays under the repo’s per-file line cap. Strictly ansible-free (stdlib only) so the runner-host CLI path keeps importing it without ansible installed.

utils.cache.schema_credentials.extract_default_credentials(creds_node: Any) → dict[str, Any]

Return the subset of a credentials: tree’s leaves that carry a literal default: Jinja string.

The shape mirrors the schema tree: nested keys stay nested. The literal string is preserved verbatim, with no rendering and no validation. Leaves WITHOUT default: are intentionally absent so the inventory’s apply_schema-generated values win the merge.

utils.cache.yaml module

In-process YAML loader with per-path memoisation.

Many CLI tools walk the same roles/<role>/<config|vars|meta>/…yml files multiple times during a single invocation (dependency resolution, service resolution, role introspection, …). Each direct yaml.safe_load(path.read_text()) re-parses YAML and re-touches disk.

This module exposes a single cached load_yaml(path) so each absolute path is parsed at most once per Python process. dump_yaml writes through and evicts the cached entry. invalidate(path) is the explicit escape hatch when a caller knows a file was changed externally.

CACHE SEMANTICS - Cache key: (str(Path(path).resolve()), st_mtime_ns, st_size).

Symlinks resolve once at the cache boundary so the same target shares an entry regardless of how callers spell the path. The mtime+size signature invalidates entries whenever the file is rewritten by an external process (e.g. Ansible’s copy: module persisting tokens while the same Python process keeps running across many calls).

• Cached value: the list of documents yaml.safe_load_all returns for that file. Single-document YAML files cache as a 1-element list, so a callsite that asks for the single doc (load_yaml / load_yaml_any) and a callsite that asks for the multi-doc list (load_yaml_all) share the same parse. The same list / dict instances are returned to every caller, so callers that mutate the result MUST copy.deepcopy() it first. Treat the value as read-only.

• Lifetime: process-wide. CLI tools that never rewrite their inputs see the same hit rate as before; long-lived Ansible processes pick up out-of-band writes on the very next read.

utils.cache.yaml.dump_yaml(path, data: Any) → None

Write data to path as YAML and evict the cached entry.

Accepts any safe-dumpable Python structure (dict, list of dicts, scalar). The next load_yaml* call for path will parse the just-written file and refill the cache, which is what callers expect when a CLI tool or test fixture edits a file mid-run.

utils.cache.yaml.dump_yaml_str(data: Any, *, sort_keys: bool = False, default_flow_style: bool = False, width: int | None = None) → str

Serialise data to a YAML string.

Use this for stdout output, log messages, or any place a YAML representation is needed without an on-disk file. There is no cache involved (the destination is a string, not a path), but the helper exists so callers don’t have to import yaml just to call yaml.safe_dump — keeping every YAML touchpoint in utils.cache.yaml.

Parameters:

• data – the structure to serialise.

• sort_keys – emit mapping keys in sorted order.

• default_flow_style – emit collections inline instead of block style.

• width – column to wrap long scalars at, None for the PyYAML default.

utils.cache.yaml.invalidate(path) → None

Drop one path from the cache.

Use when the file was changed by code that does not go through dump_yaml (e.g. an external subprocess, a manual file rewrite, a fixture in tests). With the mtime+size cache key, normal out-of-band writes invalidate automatically; this helper is the explicit escape hatch when the timestamps must not be trusted.

utils.cache.yaml.load_yaml(path, *, default_if_missing: Any = <object object>) → dict[str, Any]

Load a YAML file as a dict, memoised by absolute path.

default_if_missing controls the missing-file behaviour: - Default (_MISSING): raise FileNotFoundError for callers that

want to fail loud (resolution code).

• Pass a value (typically {}): return that value when the file does not exist, matching the behaviour expected by cli.administration.inventory.provision.yaml_io.load_yaml.

Raises TypeError when the YAML root is not a mapping; use load_yaml_any for files whose root is a list or scalar.

utils.cache.yaml.load_yaml_all(path, *, default_if_missing: Any = <object object>) → list

Memoised multi-document YAML load.

Returns the list of documents from a multi-doc YAML file (--- ... --- ...). Shares the unified per-path cache with load_yaml_any() / load_yaml(), so loading a file as multi-doc and then as single-doc (or vice versa) parses it exactly once.

utils.cache.yaml.load_yaml_all_str(text: str) → list

Parse a multi-document YAML string into a list of documents.

No caching (the source is a string, not a path) — provided for symmetry with load_yaml_str() so callers never need import yaml for the parse.

utils.cache.yaml.load_yaml_any(path, *, default_if_missing: Any = <object object>) → Any

Memoised YAML load that accepts any root shape (dict, list, scalar). Use this for files where the root is not a mapping (e.g. Ansible task lists, the root list in meta/variants.yml).

Empty files surface as {} (matching yaml.safe_load(…) or {} semantics that most call sites already rely on).

utils.cache.yaml.load_yaml_str(text: str) → Any

Parse a YAML string.

Use this when the YAML payload comes from somewhere other than a file (HTTP response body, stdin, an intermediate buffer). The path-keyed cache cannot help here — every call parses afresh — but the helper exists for symmetry with dump_yaml_str() so callers never have to import yaml directly.

Module contents

In-process caches for runtime data (applications, users, domains).

Domain modules:

• utils.cache.base — cross-cutting helpers (constants, deep-merge, signatures, render machinery).

• utils.cache.applications — variants + get_merged_applications (ansible-free at import time, runner-host friendly).

• utils.cache.users — get_user_defaults, get_merged_users.

• utils.cache.domains — get_merged_domains.

• utils.cache.yaml — process-wide load_yaml / load_yaml_any parser cache shared by everything that reads YAML off-disk.

• utils.cache.files — process-wide project-tree walk + file-content cache (read_text, iter_project_files, iter_non_ignored_files).

• utils.cache.gitignore — cached .gitignore pattern loader + path matcher used by lint/integration tests that need a “files git would track” view without spawning git.

Use the domain module directly. The package-level _reset_cache_for_tests clears every domain’s caches plus the shared fingerprint memo in one call — preferred over hand-rolling N _reset() calls per test fixture.

PROJECT_ROOT is the single source of truth for the repository root. Every callsite that needs it — CLI tools, plugins, tests — MUST import this value rather than re-derive it via Path(__file__).parents[N] or os.pardir chains; tests/lint/repository/test_project_root_import.py enforces that.