utils.roles.applications.services package¶

Submodules¶

utils.roles.applications.services.config module¶

Resolve a single services.<key>.<field> with provider fallback.

A consuming role MAY override a shared service’s config field in its own meta/services.yml services.<key> block. When it does not, the value falls back to the provider role’s native declaration — the role whose entity name is <key> (e.g. svc-net-tor for tor). This keeps provider defaults (e.g. tor.exclusive / tor.primary) in one place while every consumer stays free to override the field per app.

utils.roles.applications.services.config.resolve_service_config(applications: dict[str, Any], application_id: str, service_key: str, config_key: str, default: Any = None, *, service_registry: dict[str, Any] | None = None) → Any¶

Return services.<service_key>.<config_key> for application_id.

Resolution order:

the consuming role’s own override, when it declares config_key;
the provider role’s native value (role whose entity is service_key);
default.

Pass service_registry to reuse a registry already built by the caller (avoids rebuilding it once per lookup inside a loop).

utils.roles.applications.services.database module¶

utils.roles.applications.services.database.get_compose_services(applications: Mapping[str, Any], application_id: str) → dict[str, Any]¶

Return the role’s services map.

The materialised application payload exposes services under the bare services key (no compose.services wrapper). The function name is preserved for backwards compatibility with existing call sites.

utils.roles.applications.services.database.get_database_service_config(applications: Mapping[str, Any], application_id: str) → dict[str, Any]¶

utils.roles.applications.services.database.has_single_database_service(applications: Mapping[str, Any], application_id: str) → bool¶

utils.roles.applications.services.database.resolve_database_service_key(applications: Mapping[str, Any], application_id: str) → str¶

utils.roles.applications.services.engines module¶

Central “engine” dependencies (cache / queue / search / vector / DNS resolver).

A role may consume an engine either embedded (a sidecar inside its own stack) or central (a single pinned svc-db-* / svc-dns-* stack shared by many roles), selected per consumer with the shared flag exactly like the RDBMS (postgres / mariadb) pattern in utils.roles.applications.services.database.

See docs/architecture/central-engines.md for the rationale (engine on-disk state cannot live on NFS, so centralising it keeps the application roles unpinned and NFS-shareable in swarm).

utils.roles.applications.services.engines.central_consumers(applications: Mapping[str, Any], engine: str) → list[str]¶: Roles consuming engine as shared: true – drives on-demand enable of the central engine stack (deploy it only when at least one role needs it).

utils.roles.applications.services.engines.central_engine_needed(applications: Mapping[str, Any], engine: str) → bool¶

utils.roles.applications.services.engines.engine_service(applications: Mapping[str, Any], application_id: str, engine: str) → dict[str, Any]¶: The consumer’s services.<engine> block (or empty).

utils.roles.applications.services.engines.get_services(applications: Mapping[str, Any], application_id: str) → dict[str, Any]¶

utils.roles.applications.services.engines.is_engine_embedded(applications: Mapping[str, Any], application_id: str, engine: str) → bool¶: Consumer wants an embedded sidecar (enabled and not shared).

utils.roles.applications.services.engines.is_engine_enabled(applications: Mapping[str, Any], application_id: str, engine: str) → bool¶

utils.roles.applications.services.engines.is_engine_shared(applications: Mapping[str, Any], application_id: str, engine: str) → bool¶: Consumer wants the central instance (enabled and shared).

utils.roles.applications.services.errors module¶

exception utils.roles.applications.services.errors.ServicesResolutionError¶

Bases: RuntimeError

Raised when shared service resolution fails (e.g., missing mapping, invalid config).

utils.roles.applications.services.registry module¶

utils.roles.applications.services.registry.GROUP_GATED_CONSTANTS: tuple[str, ...] = ('LITELLM_USABLE',)¶

Group_vars constants whose own definition carries an in group_names gate, so a flag referencing one is dynamic even though its own text is not.

A constant hides the role id the substring rule looks for, so every SPOT of this kind MUST be listed here or the roles referencing it silently drop their provider from the static closure.

exception utils.roles.applications.services.registry.ServiceRegistryError¶

Bases: ValueError

Raised when role-local service discovery is invalid.

utils.roles.applications.services.registry.bucket_order_key(role_name: str, bucket: str | None) → tuple[int, str]¶

Tie-break key for a ready role: coarse bucket first, then name.

The bucket is a name-prefix-derived loader phase; it only breaks ties among roles with no pending run_after prerequisite, so run_after stays bucket-agnostic (an edge orders across any bucket).

utils.roles.applications.services.registry.build_covered_key_to_role(service_registry: dict[str, dict[str, Any]]) → dict[str, str]¶: Inverse of build_role_to_covered_keys(): for each covered service key (NOT including the primary), the role that covers it. Use build_role_to_primary_service_key() for the primary direction; this helper is intentionally limited to the covers: extension so callers can distinguish “this role IS the provider” from “this role covers this dep transitively”.

utils.roles.applications.services.registry.build_role_to_covered_keys(service_registry: dict[str, dict[str, Any]]) → dict[str, list[str]]¶: For each role, return [primary_key, *covers] — the set of service keys whose presence in a consumer’s services map counts as an explicit declaration of a dep on that role. covers comes from the primary entity’s covers: list (e.g. web-app-keycloak covers oauth2 because the per-app oauth2-proxy only makes sense when the IdP is on the host).

utils.roles.applications.services.registry.build_role_to_primary_service_key(service_registry: dict[str, dict[str, Any]]) → dict[str, str]¶

utils.roles.applications.services.registry.build_service_registry_from_applications(applications: dict[str, Any]) → dict[str, dict[str, Any]]¶

utils.roles.applications.services.registry.build_service_registry_from_roles_dir(roles_dir: Path) → dict[str, dict[str, Any]]¶

utils.roles.applications.services.registry.canonical_service_key(service_registry: dict[str, dict[str, Any]], service_key: str) → str¶

utils.roles.applications.services.registry.detect_deploy_type(role_name: str) → str¶

utils.roles.applications.services.registry.detect_service_bucket(role_name: str) → str¶

utils.roles.applications.services.registry.detect_service_channel(role_name: str) → str¶

utils.roles.applications.services.registry.discover_role_services(role_name: str, config: dict[str, Any]) → dict[str, dict[str, Any]]¶

utils.roles.applications.services.registry.equivalent_service_keys(service_registry: dict[str, dict[str, Any]], service_key: str) → list[str]¶

utils.roles.applications.services.registry.expand_service_tokens(tokens: list[str], roles_dir: Path) → list[str]¶

Normalize service reference tokens so a caller may name either the service/entity key (e.g. email, prometheus) or the provider application id (e.g. web-app-mailu, web-app-prometheus).

An application id is expanded to the service key(s) that role provides; every other token is kept verbatim. Order-preserving, de-duplicated. Used wherever the operator supplies service references (disable, include/prune closures) so both spellings resolve identically.

utils.roles.applications.services.registry.is_explicit_truth(value: Any) → bool¶

A services.<key> flag (enabled / shared) MAY be true if it is the literal Python True, any string containing the substring in group_names, or a string referencing a GROUP_GATED_CONSTANTS entry. Negated and compound predicates ("{{ 'X' not in group_names }}") match on purpose: they are true on some node shape, and the static closure is the co-deploy superset. Mirrors the substring test at tests/integration/roles/meta/services/test_dynamic_flags.py:194.

Static-analysis callers (ServicesResolver, resolve_service_dependency_roles_from_config, lint tests) MUST use this helper instead of comparing against True directly so that the dynamic-flag form keeps resolving deps at tooling time. The runtime path goes through ansible templating before reaching these checks, where the Jinja has already been rendered to a real boolean, so accepting the unrendered form is purely additive.

utils.roles.applications.services.registry.load_applications_from_roles_dir(roles_dir: Path) → dict[str, dict[str, Any]]¶

utils.roles.applications.services.registry.load_run_after_from_roles_dir(roles_dir: Path, role_name: str) → list[str]¶

utils.roles.applications.services.registry.ordered_primary_service_entries(service_registry: dict[str, dict[str, Any]], roles_dir: Path) → list[dict[str, Any]]¶

utils.roles.applications.services.registry.read_yaml_file(path: Path) → dict[str, Any]¶

utils.roles.applications.services.registry.resolve_service_dependency_roles_from_config(config: dict[str, Any], service_registry: dict[str, dict[str, Any]]) → list[str]¶

utils.roles.applications.services.registry.run_after_topological_order(nodes: Iterable[str], run_after_of: Callable[[str], Iterable[str]], tiebreak_of: Callable[[str], Any]) → list[str]¶

Order nodes so every run_after prerequisite precedes its dependents.

run_after edges order across any category; tiebreak_of only decides among nodes that are simultaneously ready (no pending prerequisite), so a coarse phase can stay the default for edge-less roles without constraining run_after itself.

Parameters:

nodes – role names to order (deduplicated internally).
run_after_of – role name -> its run_after role names.
tiebreak_of – role name -> sort key for ready-set tie-breaking.

Raises:

ServiceRegistryError – on a run_after cycle.

utils.roles.applications.services.resolver module¶

class utils.roles.applications.services.resolver.ServicesResolver(roles_root: Path, services_file: Path | None = None)¶

Bases: object

Resolve shared-provider roles transitively from role meta/services.yml.

Note: - Transitively follows provider roles by reading THEIR configs and applying the same

resolve_direct_service_roles_from_config() logic.

direct_includes_from_config(config: dict) → list[str]¶

resolve_transitively(root_role_name: str) → list[str]¶: BFS over roles discovered via service flags. Queue order is stable, results are deduped.

utils.roles.applications.services.resolver.resolve_direct_service_roles_from_config(config: dict, service_registry: dict[str, Any] | None = None) → list[str]¶: Single source of truth for “service -> provider role(s)” mapping.

utils.roles.applications.services.resources module¶

Compute the compose-resource footprint of a role and its shared dependencies.

Single source of truth for: parsing raw service-config resource scalars, evaluating the enabled/shared/container predicates, walking a role’s services (resolving template-gated shared services to the provider role’s entity via the service registry, loading each service once unless dedup=False), and aggregating the collected rows into a budget.

Used by the ressources CLI and by the variant resource-budget lint, so the collection/aggregation logic lives here once (no duplication).

utils.roles.applications.services.resources.aggregate(rows: list[dict[str, Any]], sum_fields: list[str] | None = None) → dict[str, Any]¶

utils.roles.applications.services.resources.collect_role_resources(role_name: str, applications: dict[str, dict[str, Any]], service_registry: dict[str, dict[str, Any]], visited: set, rows: list[dict[str, Any]], warnings: list[str], depth: int = 1, max_depth: int = 0, dedup: bool = True, loaded: set | None = None) → None¶

utils.roles.applications.services.sso module¶

Single-source-of-truth resolver for a role’s unified SSO surface.

Reads services.sso.{enabled,shared,flavor,oauth2.*} from the merged applications payload and derives the semantic predicates and flavor-specific sub-values that the SSO-proxy wiring, the redis-sidecar gating, and the consumer-side playwright env all need.

The flavor enum is {oidc, oauth2, saml} with oidc as the default when omitted. The is_proxy_gated predicate is the one place that encodes “the oauth2-proxy sidecar fronts this role” — every other layer MUST consult this helper instead of re-implementing the enabled AND flavor == 'oauth2' compound. The oauth2_* keys expose the gated upstream config so call sites resolve cross-role data through one named property rather than a literal lookup('config', app_id, 'services.sso.oauth2.<sub>') (which would otherwise trip the literal-protocol-lookup and lookup-config-path static-scan guards).

utils.roles.applications.services.sso.get_sso_config(applications: Mapping[str, Any], application_id: str) → dict[str, Any]¶

Resolve the SSO state for a consumer role.

Returns a dict with keys:: enabled bool — services.sso.enabled shared bool — services.sso.shared flavor str — one of {oidc, oauth2, saml}; default ‘oidc’ is_enabled bool — alias of enabled, semantic name is_proxy_gated bool — enabled AND flavor == 'oauth2' is_oidc_native bool — enabled AND flavor == 'oidc' is_saml bool — enabled AND flavor == 'saml' oauth2_origin_host str — services.sso.oauth2.origin.host or ‘’ oauth2_origin_port str — services.sso.oauth2.origin.port or ‘’ oauth2_acl dict — services.sso.oauth2.acl or {} oauth2_allowed_groups list — services.sso.oauth2.allowed_groups or [] oauth2_pass_access_token bool — services.sso.oauth2.pass_access_token,

default False: the gate hands the upstream a bearer token only where the app decodes one

The compound predicates are the documented call surface — call sites should prefer is_proxy_gated over manually combining enabled and flavor. The oauth2_* fields are the documented way to read the gated upstream config; templates that previously spelled out lookup('config', app_id, 'services.sso.oauth2.<sub>') SHOULD use lookup('sso', app_id, 'oauth2_<sub>') instead so the Python-side helper and the Ansible-side lookup share one schema.

utils.roles.applications.services.sso.is_potentially_enabled(value: Any) → bool¶

Return True unless the value is statically known to be disabled.

Jinja templates render at runtime, so any unrendered Jinja string (e.g. "{{ 'web-app-keycloak' in group_names }}") MUST be treated as potentially truthy by static-analysis callers (lint tests, contract checks).

Statically disabled means: the value is literal False, None, or the string literal "false" (case-insensitive, whitespace trimmed). Everything else is assumed to possibly render truthy.

Use this for static lint checks that walk meta/services.yml text. Runtime callers should prefer get_sso_config’s is_enabled predicate, which already runs against the merged + rendered payload.

utils.roles.applications.services.sso.logout_url(oidc_logout_url: str, is_proxy_gated: bool) → str¶

Return the logout target of an application’s own logout control.