plugins.lookup package¶

Submodules¶

plugins.lookup.active_alertmanager_channels module¶

class plugins.lookup.active_alertmanager_channels.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Return a sorted list of communication-channel app IDs that are deployed on this host.

Deployment check : app ID must appear in group_names. Channel check : app must declare services.prometheus.communication.channel: true

in its own role config — the self-declaration pattern (SPOT per app, no hardcoded list anywhere).

Usage in a template:: {% set _comm_channels = lookup(‘active_alertmanager_channels’) %}

‘applications’ is obtained via lookup(‘applications’), the merged-config SPOT.

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[str]]¶

When the playbook specifies a lookup, this method is run. The arguments to the lookup become the arguments to this method. One additional keyword argument named variables is added to the method call. It contains the variables available to ansible at the time the lookup is templated. For instance:

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Lookup plugins can be used within playbooks for looping. When this happens, the first argument is a list containing the terms. Lookup plugins can also be called from within playbooks to return their values into a variable or parameter. If the user passes a string in this case, it is converted into a list.

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.addon_env_flags module¶

class plugins.lookup.addon_env_flags.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘addon_env_flags’, application_id)

Single source of truth for the per-addon enable flags consumed by the Playwright addon-gating helper (tests/addon-gating.js -> skipUnlessAddonEnabled). Replaces the error-prone per-template Jinja loop: it resolves the materialised addons mapping for application_id and returns one <ADDON_ID>_ADDON_ENABLED=<true|false> line per declared addon (sorted by env key, newline-joined), where the env-key derivation matches addon-gating.js exactly.

A flag is true only when the addon is both enabled AND required. Optional addons (required: false) are intentionally skipped by the spec suite — they are a variant axis, not a guaranteed surface — so their gate flag is false even when enabled.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.addon_env_flags.env_key(addon_id: str) → str¶: Mirror addon-gating.js envKey(): upper-case, non-alphanumeric runs -> ‘_’.

plugins.lookup.alert_rule_roles module¶

class plugins.lookup.alert_rule_roles.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Return a sorted list of role IDs whose Prometheus alert rules belong in this deploy: every role that ships a template at roles/<role_id>/templates/prometheus/alert_rules.yml.j2 and is actually scraped, i.e. appears in native_metrics_apps.

Keeps alert rules at the role that owns the metric instead of collecting them in web-app-prometheus: a role declares what “unhealthy” means for itself, and the Prometheus role only concatenates what the play enabled.

Filtering on activation here rather than inside each fragment is what keeps the fragments free of boilerplate. A rule that fires on a series nothing scrapes is worse than no rule: it alerts on absent() forever.

Usage in a template:: {% for role_id in lookup(‘alert_rule_roles’) %} {% include ‘roles/’ + role_id + ‘/templates/prometheus/alert_rules.yml.j2’ %} {% endfor %}

The rendered fragment MUST start at the groups: item level (two-space indented - name: <group> entries), because it is concatenated into the single groups: list of the Prometheus rule file.

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[str]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.api module¶

class plugins.lookup.api.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘api’, ‘<provider>.<key>’) e.g. lookup(‘api’, ‘github.client_id’)

Single point of truth for proprietary / external API credentials. Resolves a dotted path against the effective API mapping: the declaration in group_vars/all/18_api.yml with the calling scope’s override merged over it. Routing every access through this plugin keeps one place to change how proprietary API credentials are sourced (vault, env, secret manager, a different provider abstraction) without touching the call sites.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.api_enabled module¶

class plugins.lookup.api_enabled.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘api_enabled’, ‘<provider>’) e.g. lookup(‘api_enabled’, ‘github’)

True when every credential the provider declares in group_vars/all/18_api.yml carries a value in the effective API mapping. Integrations that depend on a proprietary provider gate on this instead of spelling the emptiness test out per call site, so an added credential key extends the gate everywhere and the result is a real boolean rather than a string a filter has to coerce.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.application_closure module¶

Lookup application_closure: roles reachable from a seed application set.

Walks the same edges the constructor’s role set is built from, so a whitelist-scoped deploy and the full-inventory deploy agree on what a role depends on. The traversal itself lives in utils/roles/applications/in_group_deps; this only exposes it to Ansible.

{{ lookup(‘application_closure’, [‘web-app-snipe-it’]) }}

Returns the seed roles plus every role reachable through meta/main.yml dependencies and declared service edges, transitively. An empty seed yields an empty list, which callers read as “no scoping requested”.

class plugins.lookup.application_closure.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[str]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.application_gid module¶

class plugins.lookup.application_gid.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.application_gid.compute_application_gid(application_id, roles_dir='roles', base_gid=10000)¶

Pure-Python GID resolver — no ansible dependency.

Sorts every <roles_dir>/<role>/meta/services.yml-bearing role alphabetically by <role> and returns base_gid + index for the requested application_id. Extracted so callers that only need the GID computation (e.g. utils.cache.applications._build_variants on the GitHub Actions runner host, where the runner Python ships without ansible) can import THIS function instead of the LookupModule class — that one transitively pulls ansible.plugins.lookup.LookupBase and raises ModuleNotFoundError on ansible-less hosts.

An “application role” carries at least one project-owned meta/<topic>.yml marker file (services, server, rbac, volumes, schema, users).

Raises ValueError (not AnsibleError) for portability.

plugins.lookup.applications module¶

class plugins.lookup.applications.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

Terms:

full merged mapping
application_id
application_id, default value

Keyword Arguments:

roles_dir – roles tree to build from (defaults to the repository’s)
carrier – with no terms, return {"key": [...], "applications": {...}} for the constructor to park in _INFINITO_APPLICATIONS_RENDERED

plugins.lookup.applications_current_play module¶

class plugins.lookup.applications_current_play.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Return the current play application mapping using the shared resolver from utils/roles/applications/in_group_deps.

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[dict[str, Any]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.asset module¶

class plugins.lookup.asset.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘asset’, application_id, package, path_within_package) }}

Resolves a browser-facing URL for a frontend dependency declared in the role’s own package-lock.json, mirroring jsdelivr’s npm path scheme so the internal and external forms differ only by host:

internal (web-svc-cdn deployed, flavor internal):
cdn.<domain>/npm/<package>@<locked-version>/<path>

external (no web-svc-cdn, or flavor external):
cdn.jsdelivr.net/npm/<package>@<locked-version>/<path>

The version is read from roles/<application_id>/files/package-lock.json, so the URL is always the exact pinned, lockfile-reproducible version.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.asset.resolve_host(variables, loader, templar) → str¶: The asset host for the current deployment: the internal CDN origin when web-svc-cdn is deployed with flavor internal, else jsdelivr.

plugins.lookup.asset_host module¶

class plugins.lookup.asset_host.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘asset_host’) }}

Returns the origin that lookup(‘asset’, …) resolves against for the current deployment (the internal CDN origin with web-svc-cdn flavor internal, else https://cdn.jsdelivr.net). Use it to derive the CSP script-/style-src origin from the same flavor decision the asset URLs use, so the whitelist never drifts from where the assets load.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.bounded_name module¶

class plugins.lookup.bounded_name.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘bounded_name’, application_id, max_length) }}

An application’s canonical primary domain when it has at most max_length characters, otherwise its short entity name (e.g. keycloak).

Parameters:

application_id – the application whose domain names the value.
max_length – the longest value the consumer accepts; a positive integer.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.ca_injected module¶

Lookup ca_injected: whether the self-signed root CA is mounted into an application’s containers.

{{ lookup(‘ca_injected’, application_id) }}

The CA override that bind-mounts CA_TRUST.inject_cert_container is emitted under three conditions, and anything that reads the mounted file must ask the same three or it will point at a path that was never mounted. The conditions lived as four hand-copied expressions - both sys-svc-compose handlers, sys-svc-compose-ca/vars/main.yml and the msmtp template - which is one copy per place that can drift.

Returns a plain bool, so callers read as a predicate:

when: lookup(‘ca_injected’, application_id) {{ CA_TRUST.inject_cert_container if lookup(‘ca_injected’, application_id) else … }}

class plugins.lookup.ca_injected.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[bool]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.canonical_url module¶

class plugins.lookup.canonical_url.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.cdn module¶

class plugins.lookup.cdn.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.cert module¶

class plugins.lookup.cert.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.colorscheme module¶

class plugins.lookup.colorscheme.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.command_path module¶

class plugins.lookup.command_path.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Resolve executable paths from command names.

Usage:: {{ lookup(‘command_path’, ‘baudolo’) }} {{ lookup(‘command_path’, ‘baudolo’, path=’/custom/bin:/usr/bin’) }}

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_external_networks module¶

Lookup compose_external_networks: the provider role ids whose overlay the compose file of application_id references as external: true.

Usage:

{{ lookup(‘compose_external_networks’) }}

Returns a list of role ids (e.g. svc-db-redis), matching the external: true entries compose_networks emits. The swarm deploy handler pre-creates those overlays so a shared provider’s network is present even when the provider role did not run in the same play.

class plugins.lookup.compose_external_networks.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_file_args module¶

class plugins.lookup.compose_file_args.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_networks module¶

Lookup compose_networks: emit the top-level networks: block for a compose file. Reads the service_registry, the rendering role’s application_id and DEPLOYMENT_MODE from variables.

Usage in any compose.yml.j2:

{{ lookup(‘compose_networks’) }}

The output starts at column 0 (no caller-side indent).

class plugins.lookup.compose_networks.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_only module¶

Lookup compose_only: emit a key: “<value>” YAML pair only in compose mode; emit nothing under swarm.

A single SPOT for every service-level YAML key that is structurally incompatible with docker stack deploy and must therefore be omitted when DEPLOYMENT_MODE == ‘swarm’. Currently used for:

container_name — swarm rejects it alongside deploy.replicas > 1 (“can’t set container_name and X as container name must be unique”).
pull_policy — swarm rejects it as “Additional property pull_policy is not allowed” (compose-only extension).

Both call sites look identical:

{{ lookup(‘compose_only’, ‘container_name’, MY_CONTAINER) }} {{ lookup(‘compose_only’, ‘pull_policy’, ‘never’) }}

The value is always double-quoted in the output. That makes the emit safe for YAML edge cases like restart: no (bare no would otherwise parse as a boolean) without burdening the caller with quoting rules.

class plugins.lookup.compose_only.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_replicas module¶

Lookup compose_replicas: emit replicas: <N> only in swarm mode.

Single SPOT for swarm replica calculation across the repo.

Default value: len(groups[application_id]) - one task per host in the application’s inventory group. Single-host inventories yield 1; multi- host swarm inventories yield N matching the topology.

Override: pass an explicit value as the first term.

Examples

{{ lookup(‘compose_replicas’) }} # topology default {{ lookup(‘compose_replicas’, 1) }} # force replicas: 1 {{ lookup(‘compose_replicas’, N) }} # force replicas: N

In compose mode the lookup returns the empty string because docker-compose ignores deploy.replicas. Wrapping the emit in the deploy: block remains the caller’s responsibility - typically roles/sys-svc-container/templates/deploy.yml.j2.

class plugins.lookup.compose_replicas.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_restart module¶

Lookup compose_restart: emit restart: <policy> only in compose mode.

In swarm mode the restart: top-level service key is silently ignored because swarm replaces it with deploy.restart_policy (rendered by roles/sys-svc-container/templates/deploy.yml.j2). Emitting restart: under swarm produces an “Ignoring unsupported options” warning at docker stack deploy time and double-declares the intent.

Default call site:

{{ lookup(‘compose_restart’) }}

resolves to the global DOCKER_RESTART_POLICY from group_vars/all/00_general.yml. That covers every caller whose intent is “use the operator-wide default policy”.

When a template needs a per-call override (either via Jinja-scope {% set docker_restart_policy = '...' %} or a hardcoded literal), the caller passes the policy explicitly:

{{ lookup(‘compose_restart’, docker_restart_policy | default(DOCKER_RESTART_POLICY)) }} {{ lookup(‘compose_restart’, ‘on-failure’) }}

The override path lives at the caller because Jinja-scope {% set %} bindings are visible only in the rendering template, never in the Ansible variable context that the lookup sees.

class plugins.lookup.compose_restart.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_volumes module¶

Lookup compose_volumes: render the top-level volumes: / configs: / secrets: block. Auto-wires application_id, the applications registry, DEPLOYMENT_MODE, and the storage mapping from the templating context.

class plugins.lookup.compose_volumes.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.compose_volumes.compose_volumes(applications: dict[str, Any], application_id: str, *, extra_volumes: dict[str, dict[str, Any]] | None = None, extra_configs: dict[str, dict[str, Any]] | None = None, extra_secrets: dict[str, dict[str, Any]] | None = None, deployment_mode: str = 'compose', storage: dict[str, Any] | None = None, dir_var_lib: str, render_jinja: Any = None) → str¶

Render the top-level volumes: / configs: / secrets: block.

Reads the canonical dict-of-dicts shape from roles/<role>/meta/volumes.yml (YAML key = semantic short name; optional name: field = docker volume name). Entries with type: config / type: secret emit their respective top-level sections; everything else flows through the volumes section.

A type: volume entry may set nfs: false to opt out of the swarm NFS rewrite: the volume then stays a plain node-local named volume even when storage.backend is nfs (e.g. gitaly repository storage, which upstream forbids on NFS). Any other nfs value (absent, true, or the uid/gid/mode dict consumed by the NFS subdir pre-creation) keeps the rewrite.

plugins.lookup.config module¶

class plugins.lookup.config.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘config’, application_id, config_path[, default])

applications are resolved via lookup(‘applications’)
default behavior is strict=True (missing keys raise)
if a 3rd argument (default) is provided, strict=False and that default is returned
parameters:
1. application_id
2. config_path
3. optional default value

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container module¶

class plugins.lookup.container.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘container’, application_id[, path])

path can be:

‘directories’ / ‘files’ (returns sub-dict)
‘directories.instance’ (returns leaf)

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_address module¶

Lookup container_address: addressable form of a container for docker exec, mode-aware.

Single SPOT for the swarm/compose split on container addressing.

Returns a string safe to embed unchanged into a shell container exec command:

Compose mode - the bare service-key (e.g. mattermost). Compose sets container_name: explicitly so the name resolves locally.
Swarm mode - a shell subshell that resolves the running task’s container ID at exec time, e.g. “$(/usr/bin/resolve-container-id mattermost mattermost)”. Swarm names every service <stack>_<key> and accepts no prefix matching; the helper composes that name from STACK and SERVICE_KEY args. The subshell defers resolution to the moment of exec, avoiding the stale ID and parse-time-failure pitfalls of returning a resolved value at vars-resolution time.

Because the lookup emits $(...) shell syntax, callers MUST use the ansible.builtin.shell module (not command) for the surrounding container exec invocation. A companion lint rule enforces this.

The helper script BIN_RESOLVE_CONTAINER_ID is deployed by sys-svc-compose during the constructor stage. It refuses to run on non-manager nodes (exit 64) and reports clear errors when the service has no running task (exit 65) or the task is not yet bound to a container (exit 66).

Examples

# roles/web-app-X/vars/main.yml X_CONTAINER_ADDRESS: “{{ lookup(‘container_address’, application_id, ‘x’) }}”

# roles/web-app-X/tasks/01_setup.yml - shell: |

container exec -i {{ X_CONTAINER_ADDRESS }} my-cli command

Both terms (application_id, service_key) are required. services.<service_key>.name must exist in the resolved application config; the stack name is derived from application_id via the get_entity_name filter (matching how docker stack deploy names the stack in sys-svc-compose/handlers/main.yml).

class plugins.lookup.container_address.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_depends_on module¶

class plugins.lookup.container_depends_on.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_deploy module¶

Lookup container_deploy: emit the swarm deploy: head — the replica count plus, for a role whose primary entity declares one, the placement constraint. Renders nothing outside swarm mode, so callers need no mode guard of their own.

{{ lookup(‘container_deploy’) | indent(4) }} # topology default {{ lookup(‘container_deploy’, replicas=1) | indent(4) }} # single-writer sidecar {{ lookup(‘container_deploy’, replicas=n) | indent(4) }} # explicit; ‘’ = default

Callers that also need update_config, restart_policy or resources append them under the same key.

class plugins.lookup.container_deploy.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_dns module¶

class plugins.lookup.container_dns.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘container_dns’) }}

Resolver list for the docker daemon’s dns key. Takes no terms.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_dns.bridge_address(variables: dict[str, Any]) → str¶

Return the docker bridge gateway containers can reach the host on.

Parameters:: variables – the variables in scope for the lookup.

ansible_facts only carries docker0 once the daemon has run, and the daemon config is written before that on a fresh node - facts are gathered at play start, and nothing re-gathers them after the install. Reading the fact alone therefore yields nothing on the very run that has to emit the address, which leaves the inner containers pointed at the clearnet resolver and the DinD DNS check failing on a domain only the node dnsmasq knows.

default-address-pools is where docker takes that address from, so the declaration answers before the daemon exists and answers identically afterwards. That also keeps daemon.json byte-identical across passes, which the fact-derived value could not: it appeared only from the second run on and restarted the container runtime under a live stack.

The fact is not merely late, it is wrong while it is late. Facts are read before the daemon config is applied, so docker0 still carries docker’s own default; writing the pools then moves the bridge underneath it. A swarm node measured 172.17.0.1 in daemon.json against a live 10.208.0.1, which is why the declaration answers here and the fact never does.

The pool’s size bounds the slice docker hands the bridge, so the gateway is derived from that slice rather than from the enclosing supernet - they agree only while the pool starts on a slice boundary.

plugins.lookup.container_dns.public_resolver(variables: dict[str, Any]) → str¶

Return the resolver to fall back on when the bridge may serve nothing.

Parameters:: variables – the variables in scope for the lookup.

Read strictly: a missing declaration must fail the render rather than emit a hardcoded address, which the repository forbids and lints for. Only the first entry is taken because a resolver list is capped at three and the bridge already holds one of those slots.

plugins.lookup.container_dns.render(value: Any, templar: Any) → str¶

Return an inventory value with its Jinja resolved.

Parameters:

value – whatever the inventory holds for the field.
templar – the lookup’s templar, or None outside a play.

networks.internet.dns arrives through -e @inventories/... and is read straight out of variables, which hands back whatever the file contains. While that was a literal address nobody noticed; an expression there reached the daemon as its own source text and docker refused the config with ParseAddr("{{ ... }}").

str() drops the TrustedAsTemplate tag the loader attached, and the templar returns an untagged string unrendered. Re-tagging before templating is what every sibling lookup does; without it this function would trade the daemon’s parse error for its own.

plugins.lookup.container_dns.resolve_container_dns(variables: dict[str, Any], templar: Any = None) → list[str]¶

Return the resolver list for the container runtime, most specific first.

On a Tor node the node dnsmasq resolves .onion through Tor’s DNSPort and everything else through the clearnet upstream, which is what makes per-app proxy settings unnecessary. It is only reachable on the docker bridge address because docker discards loopback resolvers, and it is absent before docker itself is installed, hence the guard.

The same reachability problem exists without Tor wherever the host resolves through a listener of its own on loopback, which is how the simulated cluster nodes are wired. Handing containers the declared clearnet address there was what a literal bridge address in the inventory used to paper over, and that literal went stale the moment the daemon’s address pool moved the bridge off Docker’s default.

Both deploy modes get the bridge: a Tor node runs the same dnsmasq listener in compose and in swarm, so both resolve onions alike.

The clearnet resolver stays as a second entry so that losing dnsmasq costs onion names rather than all name resolution. Duplicates are dropped: where the bridge is the clearnet resolver, emitting it twice makes daemon.json differ between a run before and a run after docker0 exists, and that difference restarts the container runtime underneath a live stack.

Whether a resolver answers on the bridge is not something the host’s own resolv.conf can say - a systemd-resolved stub and a node dnsmasq both read as loopback-only, and only the second one also binds the bridge. Until that is declared rather than inferred, the bridge may be an address nothing serves, so it never goes out alone: a list that would carry it by itself gets the project’s public resolver behind it. Without that the daemon has a single dead nameserver and every container and image build loses DNS.

plugins.lookup.container_extra_hosts module¶

Lookup container_extra_hosts: the single emitter of extra_hosts:.

Compose rejects a mapping that carries the key twice, so nothing else in the repo may write extra_hosts: by hand. Everything a service needs pinned goes through here, from two sources that are both optional:

The SSO back-channel pin, added automatically. On an onion deployment the OIDC endpoints handed to an application are public .onion authorities. That is right for the browser, which reaches them through Tor, but unroutable for the application’s own server-side calls (code-to-token exchange, JWKS, userinfo): a container resolves .onion through Tor’s resolver, not through DNS. Pinning the provider to a reachable address sends those calls over the local network to the proxy that already serves the vhost. Skipped unless the application has SSO enabled and the provider vhost is an .onion name, so it is a no-op on clearnet.
host_alias=true, which pins host.docker.internal at whatever reaches the host in the effective mode. A container that must talk to a service the host publishes – msmtp relaying through the swarm ingress on :587, for instance – needs the alias, and it must not hardcode host-gateway because that breaks under swarm for the reason below.
extra_hosts=, whatever the call site needs on top: a peer’s overlay VIP, or a partner service reachable only on the host.

Entries are "name:address" strings and are deduplicated in first-seen order. With nothing to emit the lookup returns an empty string.

entries_only=true returns those strings as a list instead of the rendered block, for a container that is not described by a compose file at all: web-app-discourse is built by discourse_docker’s launcher, whose only argument surface is docker_args: in its app.yml, so it needs --add-host=<entry> lines.

The automatic pin follows the effective deploy mode, and in swarm it must NOT be host-gateway: that alias resolves per node, while openresty publishes :80 with mode: host under node.role == manager, so every replica on a worker dialled its own gateway and the token exchange died with ‘Connection refused’ (7b1839681, measured on penpot in run 31340965955). Swarm therefore gets the stack host’s routable address – the destination svc-net-tor already sends all hidden-service traffic to. ansible_facts carries it because the compose file is rendered once with the manager as inventory host and swarm applies the same extra_hosts to every task.

The mode is read as compose_mode_force or DEPLOYMENT_MODE rather than the bare cluster mode, because container_networks resolves attachments through that same effective mode: a role with an override must not lose its pin and its overlay at once.

Caller-supplied entries are passed through untouched, so a call site that needs a different address per mode must still decide that itself.

Usage in any service template:

{{ lookup(‘container_extra_hosts’) | indent(4) }} {{ lookup(‘container_extra_hosts’,

extra_hosts=[‘host.docker.internal:host-gateway’]) | indent(4) }}

{{ lookup(‘container_extra_hosts’, application_id=’web-app-nextcloud’) | indent(4) }}

Usage from a non-compose call site:

{% for entry in lookup(‘container_extra_hosts’,

entries_only=True, wantlist=True) %}

–add-host={{ entry }}

{% endfor %}

class plugins.lookup.container_extra_hosts.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

Render the pins this application needs to reach onion-hosted services.

Kwargs:: application_id: role whose pins to resolve; defaults to the play var. extra_hosts: additional name:address entries from the caller. host_alias: also pin host.docker.internal to the gateway. docker_flags: emit –add-host= flags for a docker run command

line instead of a compose extra_hosts: block.

Returns a single-element list holding the rendered block, or [“”].

plugins.lookup.container_healthcheck module¶

Lookup container_healthcheck: emit a service’s healthcheck block.

Single SPOT for every container probe in the repo. The service name is the only positional argument, everything else comes from the service’s meta/services.yml entry:

myservice:

ports:

internal:
http: 8080

healthcheck:
flavor: curl path: health/ready start_period: 20m

flavor picks the probe shape (see utils/docker/healthcheck/); each flavor brings its own interval, timeout, retries and start_period, and any of those may be overridden next to it. Without a flavor the entry must carry an explicit test argv. port_key selects another entry below ports.internal for services that expose their probe on a second port.

Examples

{{ lookup(‘container_healthcheck’, service_name) | indent(4) }} {{ lookup(‘container_healthcheck’, service_name, samples=3) | indent(4) }}

class plugins.lookup.container_healthcheck.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_hostname module¶

class plugins.lookup.container_hostname.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘container_hostname’, application_id) }}

The Docker/OCI hostname for an application’s container: bounded_name at the 63-byte limit sethostname(2) enforces.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_image module¶

Lookup container_image: compose image: "<ref>" line for a container service.

Thin wrapper around the image lookup (the SPOT for resolving the <registry>/<image>:<version> reference). container_image only adds the image: "..." compose-line wrapping so templates can drop it in as a standalone statement:

# roles/web-app-X/templates/compose.yml.j2 {{ lookup(‘container_image’, application_id, ‘x’) }}

# Locally-built image (no registry upstream). {{ lookup(‘container_image’, application_id, ‘x’, custom=True) }}

All resolution kwargs (image=, version=, custom=, tag_only=) are forwarded to image. For a bare reference (no wrapping), call the image lookup directly.

Both terms (application_id, service_key) are required; they are validated by image.

class plugins.lookup.container_image.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_mail_marker module¶

Lookup container_mail_marker: path of the msmtp_curl delivery marker.

Single SPOT for the file the msmtp_curl healthcheck touches after msmtp exits 0. A deploy-time assertion that the relay accepted the startup mail has to name the same path the probe writes, and a second literal would drift the moment the probe moves it.

Usage:: {{ lookup(‘container_mail_marker’) }} -> /tmp/email_sent

class plugins.lookup.container_mail_marker.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_networks module¶

Lookup container_networks: emit the per-service networks: attachment block. Reads the service_registry, the rendering role’s application_id and DEPLOYMENT_MODE from variables.

Usage in any service template:

{{ lookup(‘container_networks’) | indent(4) }}

Pass provider_self_alias=False for a sidecar service that must not carry the provider’s own alias, and own_network_only=True for one that must not inherit the networks its provider joins as a consumer.

class plugins.lookup.container_networks.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_ports module¶

Lookup container_ports: build a compose ports: block from declared ports.

Single SPOT for every published port in the repo. A term is either the short list form or the dict form; pass one or more.

List form, unchanged:

{{ lookup('container_ports',
          ['gitea', 'http', DOCKER_BIND_HOST],
          ['gitea', 'ssh']) | indent(4) }}

produces:

ports:
  - "<DOCKER_BIND_HOST>:<services.gitea.ports.local.http>:<...internal.http>"
  - "<services.gitea.ports.public.ssh>:<services.gitea.ports.internal.ssh>"

The published (host) port is ports.local.<proto> when declared, else ports.public.<proto>; the container port is ports.internal.<proto>.

Dict form, for the shapes the list form cannot express:

{{ lookup('container_ports',
          {'service': 'coturn', 'protocol': 'stun_turn',
           'transport': ['udp', 'tcp'], 'container': 'same'},
          {'service': 'coturn', 'protocol': 'relay', 'transport': 'udp',
           'container': 'same'},
          {'service': 'openresty', 'protocol': 'http', 'mode': 'host'},
          {'service': 'funkwhale', 'protocol': 'api',
           'publish': 'ephemeral'}) | indent(4) }}

Keys:

service, protocol: Required, as in the list form.
ip: Host address to bind to; omit to publish on every interface.
transport: tcp, udp, or a list of both. One list item per transport. Omit for compose’s own default, which publishes no suffix.
container: internal (default) reads ports.internal.<proto>; same reuses the published port, for services that cannot remap (STUN/TURN, media relays).
mode: ingress (default) emits the short "host:published:container" string; host emits the long block with mode: host, which is what an edge role needs in swarm so the client address survives the hop.
publish: ephemeral publishes the container port on a host port docker picks, for a service addressed only through the compose network.

A declared value that is a {start, end} mapping renders as start-end on both sides, so a relay range needs no special term.

application_id is read from the play variables unless given as application_id=. key='expose' names the block instead of ports, for Discourse’s launcher file, whose expose: list is what becomes docker run -p there.

class plugins.lookup.container_ports.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_service module¶

Lookup container_service: swarm-style service name for queries.

Sibling of container_address for service-level commands. While container_address resolves to a runtime container ID for docker exec, this lookup returns the addressable form for container service ps, container service logs, and container ps –filter “label=com.docker.swarm.service.name=…”.

Returns a string safe to embed unchanged into a shell command:

Compose mode - the bare service-key (e.g. matomo). Compose tracks services by container_name; the bare key matches container ps -f name=matomo and is the only useful form on a compose host.
Swarm mode - <stack>_<service_key>, matching how Docker Swarm names every service and how container service ps accepts them. The stack name is derived from application_id via the get_entity_name filter, mirroring how docker stack deploy composes the stack name in sys-svc-compose/handlers/main.yml.

Both terms (application_id, service_key) are required. services.<service_key>.name must exist in the resolved application config.

class plugins.lookup.container_service.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_volumes module¶

Lookup container_volumes(service): render the per-service volumes: / configs: / secrets: block fed from meta/volumes.yml; application_id comes from the templating context. Output starts at column 0; callers add the indent (| indent(4) or whatever depth) so the block sits under the right service in compose.yml.j2.

class plugins.lookup.container_volumes.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.container_volumes.container_volumes(applications: dict[str, Any], application_id: str, service: str, *, extra_volumes: list[Any] | None = None, extra_configs: list[dict[str, Any]] | None = None, extra_secrets: list[dict[str, Any]] | None = None, render_jinja: Callable[[str], str] | None = None) → str¶: Return the per-service mount block (volumes: + configs: + secrets:) as a YAML string starting at column 0.

plugins.lookup.current_play_redirect_domains module¶

class plugins.lookup.current_play_redirect_domains.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Return CURRENT_PLAY_REDIRECT_DOMAINS as a list of {source,target} dicts.

Merges, in order:

user-supplied redirect_domain_mappings (if any)
{DOMAIN_PRIMARY -> DOMAIN_HOMEPAGE} (skipped when equal); non-empty result triggers sys-stk-front-base to auto-load web-opt-rdr-domains so openresty has an SNI match for DOMAIN_PRIMARY even when no app claims it. On onion nodes (svc-net-tor in group_names with a provisioned services.tor.node) the bare node-onion apex is scheduled as an additional source targeting the same DOMAIN_HOMEPAGE when that is an onion host; a clearnet DOMAIN_HOMEPAGE (no dashboard deployed) gets no onion apex entry, so an onion vhost never redirects to clearnet. Both entries require web-opt-rdr-domains in lookup('deployment').deployed so isolated CI variants do not schedule a redirect whose target is not served.
per-app alias-to-canonical mappings from lookup('applications_current_play'), gated by AUTO_BUILD_ALIASES.

Steps 1-2 win over conflicting step-3 entries.

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[dict[str, str]]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.database module¶

class plugins.lookup.database.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Resolve database values for a given database_consumer_id.

API (STRICT):

{{ lookup(‘database’, database_consumer_id) }}
{{ lookup(‘database’, database_consumer_id, ‘url_full’) }}

Notes

want-path is optional and MUST be the second positional argument if used
kwarg want= is NOT supported (use positional want-path)

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.depends_on module¶

Lookup depends_on: emit a mode-aware depends_on: block.

Compose accepts the long form <svc>: { condition: service_X } and docker stack deploy rejects it with Additional property condition is not allowed. The two modes therefore need different YAML shapes:

compose -> map form with explicit condition: per entry
swarm -> list form (just service names; swarm ignores depends_on ordering semantically but accepts the syntax)

Hand-writing the {% if DEPLOYMENT_MODE == 'swarm' %} … {% else %} … {% endif %} per service is repetitive and easy to get wrong (forgot gate, wrong condition default, inconsistent indent). This lookup emits the whole block and switches on DEPLOYMENT_MODE itself.

USAGE

# One dependency with an explicit condition. {{ lookup(‘depends_on’, {ERPNEXT_CONFIGURATOR_CONTAINER:

‘service_completed_successfully’}) }}

# Multiple dependencies, mixed conditions. {{ lookup(‘depends_on’, {DB_CONTAINER: ‘service_healthy’,

INIT_CONTAINER: ‘service_completed_successfully’}) }}

# Multiple dependencies, all defaulting to service_started # (the Docker Compose default when no condition is given). {{ lookup(‘depends_on’, [SVC_A, SVC_B]) }}

# Mix: dict value None means “use the default condition”. {{ lookup(‘depends_on’, {SVC_A: ‘service_healthy’, SVC_B: None}) }}

KWARGS

indent (int, default 4)
Number of spaces every line BUT the first one is prefixed with. The first line has NO leading whitespace - the caller places {{ lookup('depends_on', …) }} at the desired column in the template, and Jinja substitution turns that template-side leading whitespace into line 1’s indent. Lines 2+ then carry indent extra spaces (on top of the YAML-relative indent) so they align under line 1.

Concretely: writing `` {{ lookup(‘depends_on’, {…}) }}`` (call placed at column 4 in the template) with the default indent=4 renders as:
<prev line>
    depends_on:
      db:
        condition: service_healthy
<next line>
indent=0 collapses everything to the left edge - useful when the lookup is rendered into top-level YAML (no service wrapper).
default_condition (str, default service_started)
Condition used for entries whose value is None (dict) or for plain-name entries in a list term.

mode (str, optional)
Override DEPLOYMENT_MODE (used by unit tests). When unset, the lookup reads DEPLOYMENT_MODE from the templar variables. An unknown value falls back to compose.

EMPTY INPUT

An empty mapping or list returns the empty string so the surrounding template stays valid YAML (no orphan depends_on: keyword).

class plugins.lookup.depends_on.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.deployment module¶

class plugins.lookup.deployment.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[dict[str, list[str]]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.domain module¶

class plugins.lookup.domain.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘domain’, application_id) }}

Resolves the canonical primary domain for application_id via utils.cache.domains.get_merged_domains (cached). Per-app overrides belong in applications.<app>.domains and flow through the regular applications-merge pipeline.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.domains module¶

class plugins.lookup.domains.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘domains’) }} -> full canonical-domains map {{ lookup(‘domains’, application_id) }} -> subtree for one role {{ lookup(‘domains’, application_id, default) }} -> subtree with fallback
Behavior:: Builds (and caches) the canonical-domains map via utils.cache.domains.get_merged_domains. Per-app overrides belong in applications.<app>.domains (canonical/aliases) and flow through the regular applications-merge pipeline. No upstream set_fact required.

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.email module¶

class plugins.lookup.email.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.engine module¶

Lookup engine: resolve a central/embedded engine connection for a consumer.

{{ lookup(‘engine’, ‘redis’, application_id) }} # full dict {{ lookup(‘engine’, ‘redis’, application_id, ‘url’) }} # one field {{ lookup(‘engine’, ‘redis’, application_id, ‘session_path’) }} # phpredis session.save_path

Mirrors the database lookup but for the non-RDBMS engines registered in utils.roles.applications.services.engines.ENGINES (redis, memcached, rabbitmq, elasticsearch, typesense, qdrant, unbound). When the consumer declares services.<engine>.shared: true the connection points at the central svc-db-* / svc-dns-* stack with the consumer’s own provisioned credentials; otherwise it points at the embedded sidecar inside the consumer’s own stack.

See docs/architecture/central-engines.md.

class plugins.lookup.engine.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.engine.qdrant_consumer_key(applications: dict[str, Any], svc_id: str) → str¶

Return the engine-wide key a data-owning qdrant consumer presents.

Parameters:

applications – the resolved applications mapping.
svc_id – the qdrant service role id.

qdrant has no per-consumer principals, only two engine-wide keys, so a consumer that owns data here holds the engine’s own key. The read-only key is deliberately never returned: it belongs to the MCP adapter, whose whole read-only guarantee rests on being unable to present anything else.

plugins.lookup.engine.redis_session_path(host: str, port: int, prefix: str, username: str, password: str) → str¶

Return the phpredis session.save_path of one consumer.

Parameters:

host – redis host the consumer reaches.
port – redis port.
prefix – consumer prefix; session keys become <prefix>:session:<id>.
username – ACL user, sent only together with a password.
password – redis password; empty for an embedded sidecar.

plugins.lookup.fediwall_active module¶

Role-local lookup plugin for web-app-fediwall.

{{ lookup(‘fediwall_active’, want_path) }}

For every Mastodon-API-compatible Fediverse sibling listed under the SPOT meta/services.yml.fediwall.fediverse_siblings that is also active on the current host (i.e. its application_id is in group_names), return one entry per want_path:

siblings -> the active sibling application_ids themselves domains -> each sibling’s primary domain (canonical[0]) url_bases -> each sibling’s TLS url.base

(= the same value lookup('tls', s, 'url.base') returns)

Used by:: meta/server.yml -> connect-src CSP whitelist (url_bases) templates/wall-config.json.j2 -> default servers list (domains)

Sharing one lookup keeps the active-set logic in a single SPOT and mirrors the tls lookup’s positional want API.

class plugins.lookup.fediwall_active.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any], variables: dict | None = None, **kwargs: Any) → list[list[str]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.fediwall_active.resolve_for_want(siblings: Iterable[str] | None, group_names: Iterable[str] | None, want: str, resolver: Callable[[str], str] | None) → list[str]¶

Return one value per active sibling, depending on want.

For want == 'siblings' the active sibling names themselves are returned (no resolver call). For 'domains' and 'url_bases' the caller-supplied resolver(sibling) produces the value.

plugins.lookup.fediwall_active.select_active(siblings: Iterable[str] | None, group_names: Iterable[str] | None) → list[str]¶

Return the subset of siblings whose names appear in group_names.

Order of siblings is preserved. Duplicates in siblings are preserved as-is. None inputs are treated as empty.

plugins.lookup.group_roles module¶

Lookup group_roles: the roles of a role group in run_after order.

Returns one {'role': …, 'app': …} mapping per role, so a stage can loop over the group instead of including a generated task file per group:

loop: “{{ lookup(‘group_roles’, ‘web-app’) }}”

role is the directory under roles/, app the role’s application_id – the value the deployment whitelist and application_allowed speak in.

class plugins.lookup.group_roles.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[dict[str, str]]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.image module¶

Lookup image: fully-qualified image reference for a container service, mode-aware. The SPOT for resolving the <registry>/<image>:<version> reference.

Returns the bare reference (no image: wrapping):

Compose mode - <image>:<version> (e.g. mattermost/mattermost-team-edition:11.8.0). No registry prefix is applied because compose pulls straight from the upstream registry.
Swarm mode - the same reference, prefixed with the in-cluster registry host:port (e.g. manager.example.com:5000/mattermost/mattermost-team-edition:11.8.0) when swarm.registry.host and swarm.registry.port are set. With no registry configured, swarm falls back to the un-prefixed form.

The prefixed form is byte-identical to the container’s .Config.Image after deploy, so it can be matched against a running container exactly.

Inputs come from the merged application config (services.<key>.image and services.<key>.version). image= and version= keyword overrides bypass those values for ad-hoc references (e.g. side-car init containers that do not appear in the services map).

Locally-built images are declared once, in the service’s config: services.<key>.custom: true names the image <entity_name>_custom (where <entity_name> comes from utils.roles.entity.name.get_entity_name) shared across an app’s custom services, while a non-empty STRING names it <custom>_custom for a second, distinct custom image inside one app (where the shared <entity>_custom would collide). The compose template’s build: directive does the actual building; this lookup only derives the name the build is tagged with. A custom= kwarg overrides the config declaration, and an image= override wins over both. services.<key>.image is ignored entirely in custom mode (the lookup never reads it).

When version is unset (no services value, no override), the lookup returns the bare <image> (unpinned) and emits a display.warning so the operator notices the unpinned reference without aborting the play.

The tag_only=True kwarg returns the reference WITHOUT the swarm registry prefix. Use it when the ref must be interpolated inline against the upstream registry rather than the in-cluster mirror.

For the compose image: "<ref>" line, use the container_image lookup, which wraps this one.

Both terms (application_id, service_key) are required.

class plugins.lookup.image.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.local_mtime_qs module¶

class plugins.lookup.local_mtime_qs.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Return a cache-busting string based on the LOCAL file’s mtime.

Usage (single path → string via Jinja):: {{ lookup(‘local_mtime_qs’, ‘/path/to/file.css’) }} -> “?version=1712323456”
Options:: param (str): query parameter name (default: “version”) mode (str): “qs” (default) → returns “?<param>=<mtime>”

“epoch” → returns “<mtime>”
Multiple paths (returns list, one result per term):: {{ lookup(‘local_mtime_qs’, ‘/a.js’, ‘/b.js’, param=’v’) }}

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.mail_provider module¶

Resolve the role acting as mail provider for this deploy.

Usage:: {{ lookup(‘mail_provider’) }}

See utils.mail.provider for why this is not simply MAIL_PROVIDER.

class plugins.lookup.mail_provider.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None = None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.mcp_client_bearer module¶

Lookup mcp_client_bearer: the secret a client presents to a provider.

“{{ lookup(‘mcp_client_bearer’, ‘web-app-gitea’) }}” -> the bearer an admitted client sends, or “” when none is minted yet

Which secret that is depends on mcp.credential.source: an adapter fronts its upstream and issues its own bearer under the role’s secrets.credentials, while a native or plugin provider hands out the token its principal holds in the store. Reading the wrong one yields a secret that exists, authenticates something, and is rejected by the endpoint under test.

The choice is not repeated here: it is resolve_credential, the same function mcp_servers renders every client’s connection with, so a test built on this lookup measures the credential the deployment actually ships.

class plugins.lookup.mcp_client_bearer.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.mcp_credential module¶

Read the MCP credential a provider issued, from its declared owner.

The token lives in the store under the user named by the provider’s mcp.credential.owner, keyed by the provider’s own application_id. Spelling that out at each call site repeated the owner resolution, the two .get fallbacks and the trim in eighteen places, so a role that read it slightly differently read a different thing.

Usage:: “{{ lookup(‘mcp_credential’, ‘web-app-gitea’) }}” -> the reader token, or “” when none has been issued yet “{{ lookup(‘mcp_credential’, ‘web-app-gitea’, ‘mcp-writer’) }}” -> the writer token, stored beside it under a role-suffixed key

An empty string is a valid answer: a provider that has not minted its credential yet is the normal state early in a play, and the callers distinguish it from a wrong one by probing.

class plugins.lookup.mcp_credential.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.mcp_servers module¶

Lookup mcp_servers: which MCP servers this client may connect to.

{{ lookup(‘mcp_servers’) }} -> {“selected”: […], “rejected”: […]}

Discovery is an intersection, not a list. A provider names the clients it admits in allowed_consumers; a client names the transports and auth schemes it can present. Only a pair that satisfies both, and whose provider credential actually resolves, reaches selected.

Everything else lands in rejected with a stable code: consumer_not_allowed, transport_unsupported, auth_unsupported, credential_missing, endpoint_unreachable and tools_unenforceable.

consumer_not_allowed is a decision and simply narrows the result. transport_unsupported, auth_unsupported, endpoint_unreachable and tools_unenforceable mean the provider did authorize this consumer and the connection still cannot be rendered, so they abort the run instead of disappearing quietly: a client that silently ends up with fewer tools than the deployment declared is indistinguishable from one that works.

tools_unenforceable is that same rule applied to the tool policy rather than the connection. A client filters by tool name, so a provider that resolves no callable name gives it nothing to pin and the rendered filter would be empty. That happens two ways, and neither is a working pairing: a surface composed at runtime keeps its contract in tools.categories, which no per-tool filter can express, and a surface that withholds every tool it serves as mutating has nothing left to offer. Advertising either as a configured server would leave an agent wired to a provider it can never call.

Providers are narrowed to application_closure(deployment.whitelist), the same set sys-service-loader preloads from. Inventory membership, group_names and the raw whitelist all admit roles the run never deploys, and a client then carries an endpoint nothing is serving. In compose the closure covers everything present, so only swarm makes the difference visible.

credential_missing is the one authorized rejection that does not abort, because a provider deploying later in the same play has not written its secret yet. That exemption ends at the reconciliation stage, which runs once every provider has: assert_authorized_are_renderable(discovery, strict=True) treats it as fatal there.

The provider credential is whatever mcp.credential declares: owner names the principal, source where its secret lives (token_store or the role’s own credentials), key the entry. No provider inherits the administrator’s token.

class plugins.lookup.mcp_servers.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: Sequence[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.mcp_servers.assert_authorized_are_renderable(discovery: Mapping[str, Any], strict: bool = False) → None¶

Abort when a provider authorized this consumer but nothing can connect.

Parameters:

discovery – the {"selected": ..., "rejected": ...} result.
strict – also treat credential_missing as fatal. Its exemption buys exactly one thing, a provider that deploys later in the same play and has not written its secret yet. The reconciliation stage runs after every provider, so there the exemption would only hide a provider whose credential never resolved at all.

plugins.lookup.mcp_servers.build_mcp_discovery(servers: Sequence[Mapping[str, Any]] | None, consumer_id: str, consumer: Mapping[str, Any], credentials: Mapping[str, tuple[str, str]]) → dict[str, list[dict[str, Any]]]¶

Return the selected and rejected MCP servers for one client role.

Parameters:

servers – roles_with_service('mcp', direction='server') entries.
consumer_id – application_id of the client doing the discovery.
consumer – the client’s own mcp block, carrying supported_transports and supported_auths.
credentials – resolved (token, owner) per provider role id.

plugins.lookup.mcp_servers.endpoint_url(endpoint: Mapping[str, Any]) → str¶

Return the URL a client connects to.

Parameters:: endpoint – the discovered endpoint mapping.

Every discovered endpoint authenticates through a header. A provider that keys its session by URL segment instead cannot scope that segment per consumer or revoke it, so it is fronted by an adapter and the segment stays between the adapter and its upstream.

plugins.lookup.mcp_servers.resolve_consumer_id(vars_: Mapping[str, Any], templar: Any) → str¶

Return the application_id of the client performing the discovery.

Parameters:

vars – the variables in scope for the lookup.
templar – the templar used to render a deferred application_id.

plugins.lookup.mcp_servers.role_credentials_of(applications: Mapping[str, Any], role: str) → Mapping[str, Any]¶

Return a role’s own secrets from the merged applications payload.

Parameters:

applications – the merged applications mapping.
role – the provider’s application id.

They live at secrets.credentials, which is where the roles’ own vars and every other consumer address them. Reading the top-level credentials instead returns nothing for every provider whose credential.source is credentials, and the deploy then aborts on a secret that was provisioned all along.

plugins.lookup.mcp_servers.select_deployed(servers: list[dict[str, Any]] | None, closure: set[str] | None) → list[dict[str, Any]]¶

Drop providers this run does not deploy.

An empty or unresolvable closure filters nothing: discovery that silently dropped every provider would look exactly like a deployment without any, which is the failure this module refuses to produce elsewhere.

plugins.lookup.mesh_address module¶

Lookup mesh_address: a host’s primary mesh address, or ‘’ off the mesh.

Usage:

{{ lookup(‘mesh_address’, STACK_HOST) }}

Takes exactly one term, the inventory hostname.

class plugins.lookup.mesh_address.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.mesh_ranges module¶

Lookup mesh_ranges: the address ranges the WireGuard meshes occupy, as a list.

Usage:

{{ lookup(‘mesh_ranges’) }}

Takes no terms.

class plugins.lookup.mesh_ranges.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.minio module¶

class plugins.lookup.minio.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Resolve MinIO object-store values for a given minio_consumer_id.

Mirrors plugins/lookup/database.py: branches on the consumer’s services.minio.shared flag to yield either central (shared) or embedded (local) connection details.

API (STRICT):

{{ lookup(‘minio’, minio_consumer_id) }}
{{ lookup(‘minio’, minio_consumer_id, ‘url’) }}

Notes

want-path is optional and MUST be the second positional argument
kwarg want= is NOT supported (use positional want-path)

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.native_metrics_apps module¶

class plugins.lookup.native_metrics_apps.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Return a sorted list of deployed application IDs that satisfy all of:

services.prometheus.native_metrics.enabled: true in their role config
a prometheus.yml.j2 template at roles/<app_id>/templates/
reachable from this prometheus: NOT on a node-local force_bridge network under swarm, which a swarm prometheus can neither join nor resolve, and NOT pinned to compose by its own compose_mode_force while the cluster runs swarm, which would make the scrape target name the wrong object. This is the single point that gates unreachable apps out of the native-metrics precreate + scrape.

Used by web-app-prometheus/templates/configuration/prometheus.yml.j2 to auto-discover apps that expose a native /metrics endpoint without hardcoding each app name.

Usage in a template:: {% for app_id in lookup(‘native_metrics_apps’) %} {% include ‘roles/’ + app_id + ‘/templates/prometheus.yml.j2’ %} {% endfor %}

‘applications’ is obtained via lookup(‘applications’), the merged-config SPOT.

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[str]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.native_metrics_apps.compose_mode_force(app_id: str) → str¶

Return the role’s own deploy-mode override.

Parameters:: app_id – the role directory name.
Returns:: The literal compose_mode_force from the role’s vars, or "" when it is absent or a Jinja expression, which cannot be resolved from the scraping role’s variable scope and so leaves the override invisible here.

plugins.lookup.native_metrics_target module¶

Lookup native_metrics_target: scrape target for an app that exposes its own metrics port. Reads services.prometheus.native_metrics for the service key and port, then delegates host resolution to scrape_target so swarm and compose stay on one implementation.

Usage in a per-app prometheus.yml.j2 fragment:: targets: [“{{ lookup(‘native_metrics_target’, native_prometheus_application_id) }}”]

class plugins.lookup.native_metrics_target.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.network_driver module¶

class plugins.lookup.network_driver.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘network_driver’, probe, network_role_id) }}

Parameters:

probe – the registered result of container network ls –filter name=<entity> –format ‘{{json .}}’.
network_role_id – the role whose entity name names the network.

Returns:

The driver of the network named exactly after that entity, ‘’ when none exists.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.nfs_client_src module¶

class plugins.lookup.nfs_client_src.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

NFS client mount src (server:path): {{ lookup(‘nfs_client_src’) }}.

{{ lookup('nfs_client_src', 'controller') }} addresses the server as the Ansible controller reaches it.

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.nfs_export_base module¶

class plugins.lookup.nfs_export_base.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

NFS export base from the provider SPOT: {{ lookup(‘nfs_export_base’) }}.

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.nfs_flavor module¶

class plugins.lookup.nfs_flavor.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Resolved svc-storage-nfs-server flavor (kernel/ganesha): {{ lookup(‘nfs_flavor’) }}.

Reads the merged-config SPOT through lookup(‘config’); the flavor itself is declared in svc-storage-nfs-server/meta/services.yml.

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.nfs_fstype module¶

class plugins.lookup.nfs_fstype.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

NFS mount fstype (nfs4/nfs): {{ lookup(‘nfs_fstype’) }}.

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.nfs_mount_opts module¶

class plugins.lookup.nfs_mount_opts.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

NFS mount options: {{ lookup(‘nfs_mount_opts’) }}.

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.nfs_state_path module¶

class plugins.lookup.nfs_state_path.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Cluster-shared NFS state dir: {{ lookup(‘nfs_state_path’) }}.

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.nginx module¶

Resolve nginx path configuration and an optional domain-specific server config path under <servers_dir>/<protocol>/<domain>.conf.

The want= kwarg is intentionally ignored: want-path is ALWAYS the first positional argument so callers can’t accidentally pass it both ways.

class plugins.lookup.nginx.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.node_max_old_space_size module¶

Derive a Node.js --max-old-space-size (MB) from a service’s mem_limit.

Lookup form — the app config is resolved internally, so callers no longer pipe lookup('applications') in:

{{ lookup(‘node_max_old_space_size’, application_id, service_name) }} {{ lookup(‘node_max_old_space_size’, application_id, service_name,

pct=0.125, min_mb=256, hardcap_mb=1024) }}

Heuristics (defaults):

candidate = 35% of mem_limit
min = 768 MB (required minimum)
cap = min(3072 MB, 60% of mem_limit)

If mem_limit (container cgroup RAM) is smaller than min_mb an exception is raised, to prevent Node’s heap from exceeding the cgroup and being OOM-killed.

class plugins.lookup.node_max_old_space_size.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘node_max_old_space_size’, application_id, service_name[, pct=, min_mb=, hardcap_mb=, safety_cap_pct=])

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.node_max_old_space_size.node_max_old_space_size(applications: dict, application_id: str, service_name: str, pct: float = 0.35, min_mb: int = 768, hardcap_mb: int = 3072, safety_cap_pct: float = 0.6) → int¶

Derive Node.js –max-old-space-size (MB) from services.<service>.mem_limit.

Raises AnsibleError if mem_limit is missing/invalid OR if mem_limit (MB) < min_mb.

plugins.lookup.objstore module¶

class plugins.lookup.objstore.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Resolve engine-agnostic S3 object-store values for a consumer.

Mirrors plugins/lookup/database.py (postgres|mariadb): branches on which object-store engine (seaweedfs|minio) the consumer has enabled and yields a uniform S3 connection payload, so consumer templates stay engine-agnostic.

API (STRICT):

{{ lookup(‘objstore’, consumer_id) }}
{{ lookup(‘objstore’, consumer_id, ‘url’) }}

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.objstore_consumers module¶

Enumerate the roles that consume a given object-store engine on this host.

Returns the sorted role ids deployed on the current host whose merged applications config binds them to engine as a shared object store. The provider role is excluded. Selection is by that binding alone – no naming convention is assumed of a consumer.

class plugins.lookup.objstore_consumers.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: Sequence[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.path_absolute module¶

class plugins.lookup.path_absolute.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘path_absolute’, ‘roles/sys-svc-compose/tasks/utils/swarm/x.yml’) -> absolute path under the repository root (playbook_dir).

Single point of truth for the repo-root anchor, replacing the repeated [playbook_dir, '<rel>'] | path_join boilerplate at cross-role include/template/script/src call sites. Joins every term, so a path may be passed whole (‘roles/x/y.yml’) or split across terms when a segment is a variable (lookup('path_absolute', 'roles', system_service_role_name)).

run(terms, variables: dict[str, Any] | None = None, **kwargs) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.playwright_workers module¶

class plugins.lookup.playwright_workers.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘playwright_workers’)

Conservative, stable parallel-worker count for the Playwright e2e suite. Derived from the control host’s effective CPU and RAM, divided down because browser workers are heavy and run against the co-resident application stack, reserved against RAM, hard-capped for the application-under-test’s concurrency limits, and reduced further on CI. Always returns an int >= 1.

Tunable via kwargs: cpu_divisor, hard_cap, ci_cap, per_worker_gb, ram_fraction.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.playwright_workers.compute_workers(cpus: int, ram_gb: float, ci: bool, *, onion: bool = False, per_worker_gb: float = 1.5, ram_fraction: float = 0.5, cpu_divisor: int = 4, hard_cap: int = 6, ci_cap: int = 2, onion_factor: int = 2, onion_cap: int = 3) → int¶

Workers the host can carry, and the transport can use.

Parameters:

cpus – usable cores.
ram_gb – memory a worker may claim, taken from what is free rather than from what the machine has.
ci – whether a CI runner is executing, which lowers the ceiling.
onion – whether the suite reaches its target over Tor. A run against a .onion address waits on circuits rather than on the CPU: measured on one Nextcloud variant at equal mode and filesystem, the sidecar took 4817s over Tor against 911s on clearnet, and Tor burns no browser CPU. The CPU divisor cannot describe that phase, so the count is raised instead of derived from cores. Raised by a factor rather than freed, because every worker holds its own circuits.
onion_factor – multiplier applied to the CPU-derived count over Tor.
onion_cap – ceiling for the Tor case, independent of the CI cap.

Returns:

Worker count, never below one.

plugins.lookup.postgres_libraries module¶

Resolve postgres-extension library install recipes.

Usage:: {{ lookup(‘postgres_libraries’, [‘vector’, ‘bloom’]) }}

Takes a list of extension names and returns the subset that needs a custom library install on top of the official postgres base image (the contrib extensions bloom, pg_trgm and unaccent are already in the base — those are filtered out). Each returned entry is a dict with:

extension: the extension name (matches the SQL identifier)

name: the library/source name (used in the Dockerfile context)

apt_package: the PGDG package suffix, installed as
postgresql-<major>-<apt_package>

class plugins.lookup.postgres_libraries.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: Sequence[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.prometheus_integration_active module¶

class plugins.lookup.prometheus_integration_active.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Return True when the prometheus monitoring block should be emitted for the current application on the current host; False otherwise.

The condition satisfied:

‘web-app-prometheus’ is in group_names (prometheus is deployed on this host), AND
Either the current application IS web-app-prometheus, OR it declares prometheus as an enabled compose service dependency (services.prometheus.enabled: true).

Usage in a template:

{% if lookup(‘prometheus_integration_active’, application_id) %} …prometheus monitoring block… {% endif %}

Pass ‘application_id’ as term 0 (string). ‘applications’ is obtained via lookup(‘applications’), the merged-config SPOT.

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[bool]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.proxy_pass module¶

Lookup proxy_pass: thin wrapper emitting the mode-aware directive via utils.networks.proxy.resolve_upstream() + render_proxy_pass().

class plugins.lookup.proxy_pass.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.rbac_group_path module¶

Compute the canonical OIDC group path for an application role.

This lookup is the single sanctioned way to derive the Keycloak group path that an application role maps to. It replaces the scattered [RBAC.GROUP.NAME, …] | path_join idiom so the per-application OU hierarchy has one authoritative producer.

Usage:

“{{ lookup(‘rbac_group_path’,: application_id=’web-app-yourls’, role=’administrator’) }}”

-> “roles/web-app-yourls/administrator”

“{{ lookup(‘rbac_group_path’,: application_id=’web-app-wordpress’, role=’editor’, tenant=’blog.example’) }}”

-> “roles/web-app-wordpress/blog.example/editor”

“{{ lookup(‘rbac_group_path’,: application_id=’web-app-wordpress’, role=’network-administrator’) }}”

-> “roles/web-app-wordpress/network-administrator”

Contract: - application_id MUST refer to a role that declares an rbac: block

in its meta/services.yml. The role may be tenant-aware (rbac.tenancy.axis == “domain”) or not (default).

role MUST appear under rbac.roles.<role> or be the implicit administrator that the role-list contract auto-adds.
tenant MUST be passed for tenant-aware per-tenant roles, MUST NOT be passed for global-scope roles or non-tenant apps.
Every failure raises AnsibleError with an actionable message.

The Keycloak group tree behind these paths is materialised by a per-application group-ldap-mapper (roles/web-app-keycloak/tasks/update/05_per_app_mappers.yml), anchored at /<group_root>/<application_id> with group.name.ldap.attribute=description.

class plugins.lookup.rbac_group_path.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.redirect_uris module¶

class plugins.lookup.redirect_uris.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘redirect_uris’) }} {{ lookup(‘redirect_uris’, wildcard=’/cb’) }}

Returns the redirect URIs Keycloak must whitelist: one per domain of every application this play deploys (group_names) whose config enables SSO. Each URI’s scheme is resolved for the CONSUMER – its own server.tls.enabled over the global TLS_ENABLED, and always plaintext for an .onion – never from Keycloak’s own TLS state.

parameters:
wildcard: suffix appended to every URI (default ‘/*’) features: config paths ORed to gate an application (default services.sso.enabled) dedup: drop repeated URIs (default true)

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[str]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.redirect_uris.build_redirect_uris(domains: dict, applications: dict, tls_enabled: bool, *, wildcard: str = '/*', features: Iterable[str] = ('services.sso.enabled',), dedup: bool = True, deployed: Iterable[str] | None = None) → list[str]¶

Registered redirect URIs for every deployed SSO consumer, one per domain.

Parameters:

domains – app_id -> domain, list of domains, or nested mapping of those.
applications – merged applications view, used for feature gating and for each consumer’s server.tls.enabled override.
tls_enabled – global TLS default, overridden per consumer by resolve_enabled; an .onion domain always resolves to plaintext.
wildcard – suffix appended to every URI.
features – config paths ORed together to decide whether an app gets a URI.
dedup – drop repeated URIs, preserving first-seen order.
deployed – application ids this play deploys (group_names); an app outside it gets no URI. None registers every app in domains.

plugins.lookup.role_subnets module¶

Lookup role_subnets: the local subnets other roles declare.

Single SPOT for “which address block does role X own”. The block is declared once, in that role’s meta/networks.yml; anything that has to reason about the deployment’s own address space reads it from here instead of repeating the literal, which is how a CIDR ends up stated twice and then drifts.

Usage:

{{ lookup(‘role_subnets’, ‘web-app-bigbluebutton’) }} # [‘10.7.7.0/24’] {{ lookup(‘role_subnets’, SOME_ROLE_LIST) }} # one entry per role

Terms may be role ids or lists of them, mixed. A role that declares no local subnet raises, because a silently dropped entry would narrow whatever firewall or trust list consumes the result.

class plugins.lookup.role_subnets.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.role_templates_expanded module¶

class plugins.lookup.role_templates_expanded.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict | None = None, **kwargs)¶

Expand role_templates into one spec per template to render.

Parameters:: terms – [application_id, role_templates].
Returns:: A single-element list holding the expanded spec list, so the caller receives the list itself rather than its first element.
Raises:: AnsibleError – wrong term count, empty application_id, or an entry that is not a dict.

plugins.lookup.role_tests module¶

Lookup role_tests: read a value from a role’s meta/tests.yml.

lookup(‘role_tests’, application_id, ‘cli.timeout’, default=3600)

Terms: application_id and a dotted key path. A missing file, missing key, or non-mapping intermediate returns the default kwarg (None when not given). meta/tests.yml carries per-role test-harness settings (e.g. the CLI e2e timeout for roles whose test legitimately outlasts the uniform default).

class plugins.lookup.role_tests.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict[str, Any] | None = None, **kwargs) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.roles_with_placement module¶

Ansible lookup wrapper around utils.roles.meta_lookup.iter_roles_with_placement.

lookup(‘roles_with_placement’, ‘manager’)

class plugins.lookup.roles_with_placement.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[str]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.roles_with_service module¶

Enumerate consumer roles for a given service.

Returns [{id, canonical_domain, canonical_url, iframe}, …] for every role whose merged applications config declares services.<service>.{enabled, shared} as truthy. iframe reflects services.<service>.iframe (defaulting to enabled) so consumers can tell embeddable cards from those that must open in a new tab.

A role keeps declaring the service for inventory completeness but can opt out of this consumer-target list by setting services.<service>.scrape: false or services.<service>.track: false. scrape: false is used by 301-redirect-only vhosts that declare services.prometheus (so the role-wiring contract stays satisfied) yet never serve a request through lua-resty-prometheus, so they emit no app="<role>" label and have no scrape target. track: false is the symmetric matomo opt-out: static-file (autoindex) and 301-redirect vhosts that declare services.matomo keep the role-wiring intact but carry no user-facing HTML page worth a _paq tracker, so they are dropped from MATOMO_TARGET_ROLES_JSON and the matomo tracker e2e contract.

Kwargs:

scope: host (default) restricts to roles in group_names;: deployment restricts to roles present anywhere in groups, which is what a container-network consumer can actually reach; all returns every declaring role, deployed or not.
direction: opt-in MCP filter. When set (server/client/both),: only roles whose block declares that direction (or both) are returned, and each entry additionally carries transport, auth, auth_subject, credential (owner/source/ key), allowed_consumers, supported_transports, supported_auths and an endpoint dict (service_key, path, port taken from the referenced service’s ports.internal and only then ports.local: the entry is consumed to build a container-network URL, where a host-published port is always the wrong one). Callers that omit direction keep the original 4-key entries.

class plugins.lookup.roles_with_service.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: Sequence[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.scrape_target module¶

Lookup scrape_target: mode-aware <host>:<port> for a Prometheus scrape target. Swarm resolves tasks.<entity>_<service_key> (per-task DNS); compose uses compose_host or the service container name. port overrides the services.<service_key>.ports.internal.<port_kind> lookup.

class plugins.lookup.scrape_target.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.service module¶

class plugins.lookup.service.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Resolve a service by key or role name and return its aggregated deployment flags.

Usage:: lookup(‘service’, ‘matomo’) lookup(‘service’, ‘web-app-matomo’) # resolved via reverse mapping query(‘service’, [‘a’, ‘b’, ‘c’]) # single list term -> one applications load for many keys

A term unknown to the play-local registry (no app in the play declares it) falls back to the role-wide static registry read from roles_dir, the same source the sys-service-loader iterates. Such a key resolves with its provider role and all deployment flags computed against the play (typically all false). Only a term unknown to BOTH registries raises.

Optional kwarg:

roles_dir — roles directory for the static fallback registry: (default: <cwd>/roles, matching the service_registry lookup convention).

Reads ‘applications’ and ‘group_names’ from Ansible variables and discovers service providers from the role-local services metadata.

Returns a dict per term:

id — canonical service key (e.g. ‘matomo’) role — provider role name (e.g. ‘web-app-matomo’) enabled — True if any deployed app has services.<key>.enabled: true shared — True if any deployed app has services.<key>.shared: true required — True if any deployed app has both enabled AND shared (direct

or transitively via its own enabled service dependencies). “Required” was chosen over “needed” to express that the service is contractually required by a real consumer, not merely convenient.

local — True if any deployed app has enabled AND NOT shared, i.e.: the service runs embedded inside that app’s own compose stack rather than centrally. Mutually exclusive with the shared/required axis when enabled is True.

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[dict[str, Any]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.service_extensions module¶

Aggregate per-consumer service extensions across all running roles.

Usage:: {{ lookup(‘service_extensions’, ‘postgres’) }}

Returns {consumer_id: [extension, …]} for every role in group_names whose services.<service> block is enabled and declares a non-empty extensions list.

class plugins.lookup.service_extensions.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: Sequence[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.service_memory_mb module¶

Derive a memory parameter in MiB from the memory limit of a service’s container.

-Xmx{{ lookup(‘service_memory_mb’, application_id, ‘search’, 0.25) }}m memory_limit = {{ lookup(‘service_memory_mb’, application_id, entity_name, 0.5) }}M

The limit is resolved the way sys-svc-container resolves the container’s own mem_limit: services.<service>.mem_limit, then the role’s primary entity, then RESOURCE_MEM_LIMIT. share is the fraction of that limit the parameter may claim, so a heap, a cache or a per-request limit follows the container instead of a literal that drifts away from it.

A sidecar that a provider role ships into other stacks passes provider:

–maxmemory {{ lookup(‘service_memory_mb’, application_id, ‘redis’, 0.8, provider=’svc-db-redis’) }}mb

An application that declares no mem_limit for that service lets its container inherit the primary entity’s, which is sized for the application and not for the engine beside it. The result is then capped at the same share of the provider’s own limit. An application that declares the limit itself is not capped.

Sizes are read like Docker reads them: k, m and g are binary, a bare number is bytes. The result is whole MiB, the unit -Xmx<n>m, PHP <n>M, memcached -m, Redis <n>mb and PostgreSQL <n>MB all share.

class plugins.lookup.service_memory_mb.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘service_memory_mb’, application_id, service_name, share[, provider=role])

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.service_memory_mb.docker_bytes(size: Any) → int¶

Return a Docker memory size in bytes.

Parameters:: size – a size such as 2g, 1.5g, 512m or a byte count.
Raises:: AnsibleError – the size is not a number with an optional b/k/m/g unit.

plugins.lookup.service_memory_mb.share_mb(mem_limit: Any, share: Any) → int¶

Return share of mem_limit in whole MiB.

Parameters:

mem_limit – Docker size of the container limit, e.g. 2g or 512m.
share – fraction of the limit, above 0 and at most 1.

Raises:

AnsibleError – the share is no fraction in that range, the size is unreadable, or the result is below 1 MiB.

plugins.lookup.service_registry module¶

class plugins.lookup.service_registry.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Discover the role-local service registry.

Usage:: {{ query(‘service_registry’) | first }} {{ query(‘service_registry’, ‘ordered’) | first }}

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.shared_network_key module¶

Lookup shared_network_key: the compose networks key of the pre-created <entity> network of a role.

Docker compose stores that key in the com.docker.compose.network label of every network it owns, so a caller pre-creating the network out of band has to stamp the same value or compose will resolve the project’s networks against a claim the rendered file does not make. The key is derived from utils.networks.render, which renders the file itself, so the two cannot drift.

Examples

name: create the shared network ansible.builtin.command:
argv:
container

network

create

–label

“com.docker.compose.network={{ lookup(‘shared_network_key’, role_id) }}”

class plugins.lookup.shared_network_key.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.smtp_host module¶

class plugins.lookup.smtp_host.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘smtp_host’, email) }} {{ lookup(‘smtp_host’, email, in_container=true) }}

SMTP endpoint for the msmtp relay config. Takes the resolved email mapping. Pass in_container=true when the rendered file is mounted into a container instead of being read by the host’s own msmtp.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.smtp_host.resolve_smtp_host(variables: dict[str, Any], email: dict[str, Any], *, in_container: bool = False) → Any¶

Return the SMTP endpoint msmtp relays through.

Parameters:

variables – the play’s variable namespace.
email – the resolved lookup('email') mapping.
in_container – the rendered config is mounted INTO a container rather than read by the host’s own msmtp.

Returns:

The configured relay host, the loopback address, the container’s alias for its host, or a swarm manager.

Outside the containerised test rig the configured relay is used unchanged. Inside it that relay domain has neither DNS nor a route: a host carrying the stack itself reaches mail on loopback, while a bare infra node has nothing listening there and has to enter through the manager’s routing mesh.

in_container exists because one template serves two consumers. The host’s /root/.msmtprc is read by the host and loopback is right for it; the copy mounted at /etc/msmtprc is read inside a container, where 127.0.0.1 is that container’s own loopback and nothing listens on it. Such a service must also emit lookup('container_extra_hosts') so the alias resolves.

plugins.lookup.sso module¶

Ansible lookup: unified SSO state for a consumer role.

API (STRICT):

{{ lookup(‘sso’, application_id) }} → full dict
{{ lookup(‘sso’, application_id, ‘is_enabled’) }} → bool
{{ lookup(‘sso’, application_id, ‘is_proxy_gated’) }} → bool
{{ lookup(‘sso’, application_id, ‘is_oidc_native’) }} → bool
{{ lookup(‘sso’, application_id, ‘flavor’) }} → str
{{ lookup(‘sso’, application_id, ‘enabled’) }} → bool
{{ lookup(‘sso’, application_id, ‘shared’) }} → bool
{{ lookup(‘sso’, application_id, ‘logout_url’) }} → str, oauth2-proxy sign-out wrapping OIDC.CLIENT.LOGOUT_URL when the app is proxy-gated

Wraps utils.roles.applications.services.sso.get_sso_config so templates and tasks share one source of truth with Python callers (notably plugins/lookup/compose_volumes.py).

class plugins.lookup.sso.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any], variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.sso_oidc_plugin module¶

class plugins.lookup.sso_oidc_plugin.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘sso_oidc_plugin’)

Resolves the effective OIDC plugin flavor for the Nextcloud role.

Resolution order:

“” if services.sso.enabled is falsy (no OIDC plugin should be active when the OIDC service itself is disabled — otherwise Nextcloud would still hand off to Keycloak using a redirect_uri that the client no longer whitelists).
An explicit string value at applications[‘web-app-nextcloud’] .services.sso.oidc.plugin (inventory override).
“oidc_login” if services.ldap.enabled is truthy (pulsejet/nextcloud-oidc-login, proxy-LDAP capable).
“sociallogin” otherwise (nextcloud/sociallogin).

Mirrors the former _applications_nextcloud_oidc_flavor group_vars helper that was removed in commit 77a0e16ea.

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.sso_proxy_consumers module¶

Lookup plugin: enumerate roles whose merged config has services.sso.flavor == 'oauth2' AND services.sso.enabled truthy.

Returned as a list of role names (application_id strings), in the deterministic alphabetical order of the merged-applications keys.

Usage:: {{ lookup(‘sso_proxy_consumers’) }}

No positional terms are accepted. Predicate logic delegates to utils.roles.applications.services.sso.get_sso_config so callers share one source of truth with lookup('sso', ..., 'is_proxy_gated') and plugins/lookup/compose_volumes.py.

class plugins.lookup.sso_proxy_consumers.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.stage_groups module¶

Lookup stage_groups: the ordered role-group names that belong to a deploy stage, in intra-stage call order. The stage plays loop over it so group membership follows meta/categories.yml instead of a hardcoded list:

loop: “{{ lookup(‘stage_groups’, ‘server’) }}”

class plugins.lookup.stage_groups.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[list[str]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.subnet_address_regex module¶

Lookup subnet_address_regex: match any address in the role’s local subnet.

Single SPOT for turning networks.local.subnet into an address matcher. A swarm task carries an address on every network it joins, all of them site-local, so a service that must bind to one of them needs this rather than a generic “site-local” selector.

Example

{{ lookup(‘subnet_address_regex’) }} # 192.168.24.d+

Raises when the subnet is missing or its prefix is not octet-aligned, so a topology this cannot express fails at render time instead of binding silently to the wrong network.

class plugins.lookup.subnet_address_regex.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.subnet_gateway module¶

Lookup subnet_gateway: the gateway address of the role’s local subnet.

Single SPOT for the address Docker assigns as the gateway of a role network, used where a container has to reach the host side of its own bridge.

Example

{{ lookup(‘subnet_gateway’) }} # 192.168.24.1

Raises when the subnet is missing or holds no gateway, so an unusable network fails at render time instead of reaching Docker as a malformed address.

class plugins.lookup.subnet_gateway.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.timeout module¶

Scale a base timeout (seconds) by the global TIMEOUT_FACTOR.

Replaces a bare timeout: <seconds> task keyword or module argument so every wait in the codebase scales from one knob: set TIMEOUT_FACTOR: 2 in group_vars on a slow uplink and every timeout doubles.

timeout: “{{ lookup(‘timeout’, 3600) | int }}” timeout: “{{ lookup(‘timeout’, some_base_var) | int }}”

TIMEOUT_FACTOR is read from the play variables and defaults to 1 (no scaling) when unset, so the lookup is a value-identical drop-in for the literal. An explicit factor= keyword overrides the global for one call.

class plugins.lookup.timeout.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

lookup(‘timeout’, base_seconds[, factor=TIMEOUT_FACTOR])

run(terms, variables: dict[str, Any] | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.timeout.scaled_timeout(base: Any, factor: Any = 1) → int¶

Return round(base * factor) as an int.

Parameters:

base – Base timeout in seconds (number or numeric string, >= 0).
factor – Multiplier (number or numeric string, >= 0); TIMEOUT_FACTOR.

Raises:

AnsibleError – base or factor is non-numeric or negative.

plugins.lookup.tls module¶

class plugins.lookup.tls.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables: dict | None = None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.tor_ports module¶

class plugins.lookup.tor_ports.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘tor_ports’) }}

Returns the HiddenServicePort mappings for the roles in the current deploy (group_names): a list of {'onion_port': <port>, 'target': '127.0.0.1:<port>'} dicts, sorted by port. Two opt-ins are unioned:

every port a service names under ports.onion in meta/services.yml, and

every port of a service that opts in with exposed: true.

Both are read from the variant-merged applications view, so a variant that drops the opt-in drops the forward with it. The published port binds the host interface, so the loopback target reaches it from svc-net-tor’s host-network container. Composed with the flagged forwards by lookup('tor_extra_ports') and rendered as HiddenServicePort lines in svc-net-tor’s torrc.

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.tor_ports.UDP_ONLY_CATEGORIES = frozenset({'media', 'relay', 'stun_turn', 'stun_turn_tls'})¶

Categories a hidden service cannot carry.

relay and media are UDP. stun_turn and its TLS variant do speak TCP, but the allocation they hand out is a UDP relay address the onion client can never receive on, so forwarding the signalling port buys nothing.

plugins.lookup.tor_ports.collect_exposed_ports(applications: dict[str, Any], deployed_roles: list[str]) → list[int]¶

Single-int TCP ports of every deployed service flagged exposed: true in the variant-merged applications view, sorted.

exposed is an explicit per-service opt-in (default false) that makes the service reachable over the node onion: its port gets a dedicated HiddenServicePort. Because applications is the variant-merged config (base meta/services.yml deep-merged with the round’s meta/variants.yml override), a service is forwarded only in the variant/config where it sets exposed: true — which is exactly what lets the two DB variants be tested over Tor (v0 exposed -> reachable, v1 not -> refused).

plugins.lookup.tor_ports.collect_onion_ports(applications: dict[str, Any], deployed_roles: list[str]) → list[int]¶

Single-int TCP ports a deployed role asks for by name under ports.onion.

ports.onion is a category-keyed map of booleans alongside internal / local / public: it names which of a service’s already-declared ports should answer on the node onion, and nothing else. The port number comes from ports.local when that category is declared there and from ports.public otherwise, which is the precedence container_ports publishes by, so the forward always targets the port the service is actually reachable on.

Naming the category is what makes this safe. Sweeping ports.public wholesale forwards ports the deployment never publishes – ldaps 636 is declared next to network.public: false, and Mailu’s implicit-TLS ports are dropped from the publish list whenever TLS is off, which an onion deployment always is.

A service that only covers: a capability another deployed role provides: is skipped for the same reason: it is a legacy co-tenant, the provider owns the host ports, and forwarding the co-tenant’s onion ports advertises endpoints nothing answers on. The Mailu-to-Stalwart migration round deploys both, and the ports Mailu alone would serve (110, 143, 587) are dead there.

plugins.lookup.tor_reachable module¶

class plugins.lookup.tor_reachable.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘tor_reachable’, application_id) }}

Returns whether application_id is reachable over the node onion in the current (variant-merged) applications view – see is_reachable.

parameters:
1. application_id

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[bool]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.tor_reachable.is_reachable(applications: dict[str, Any], application_id: str) → bool¶

True when application_id is reachable over the node onion, i.e. when svc-net-tor is deployed for it (services.tor.enabled) AND at least one of its services opts in with exposed: true, which is what earns a HiddenServicePort (collect_exposed_ports).

Both halves are required: tor.enabled is the dependency edge, not the port forward, so a client picking the Tor transport from that flag alone dials a port with no HiddenServicePort and no host publish.

plugins.lookup.tor_socks module¶

class plugins.lookup.tor_socks.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘tor_socks’) }} -> 127.0.0.1:<port> {{ lookup(‘tor_socks’, ‘host.docker.internal’) }} -> host.docker.internal:<port> {{ lookup(‘tor_socks’, ‘0.0.0.0’) }} -> 0.0.0.0:<port>

Single spot for the node Tor SOCKS endpoint so 127.0.0.1:<port> and its siblings are never hand-composed: the port comes from svc-net-tor services.tor.ports.local.socks and callers only pass the host (default 127.0.0.1).

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.tor_socks.resolve_socks_endpoint(applications: dict[str, Any], host: str) → str¶

Return <host>:<port> for the node Tor SOCKS proxy.

The port is the single source of truth svc-net-tor services.tor.ports.local.socks; only the host varies by caller context (loopback for host-network consumers, host.docker.internal for a bridged sibling container, 0.0.0.0 for the daemon bind).

plugins.lookup.tor_socks_proxy module¶

class plugins.lookup.tor_socks_proxy.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:

{{ lookup(‘tor_socks_proxy’) }} -> socks5h://tor:<port> (swarm): -> socks5h://host.docker.internal:<port> (compose)

Single spot for the outbound Tor SOCKS proxy URL of a containerised client, so the swarm/compose split is not restated per consuming role.

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[str]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.tor_socks_proxy.resolve_proxy_url(applications: dict[str, Any], deployment_mode: str) → str¶: Return the socks5h:// URL a sibling container uses to reach Tor: the tor overlay alias in swarm, the host gateway in compose.

plugins.lookup.unit_name module¶

class plugins.lookup.unit_name.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Lookup plugin entrypoint.

Usage:

“{{ lookup(‘unit_name’, system_service_id) }}”
“{{ lookup(‘unit_name’, system_service_id, suffix=’.timer’) }}”
“{{ lookup(‘unit_name’, system_service_id, suffix=False) }}” # no suffix

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.users module¶

class plugins.lookup.users.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[Any]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.users_non_reserved module¶

class plugins.lookup.users_non_reserved.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

Usage:: {{ lookup(‘users_non_reserved’) }}

Returns the users an identity directory provisions – those carrying identity in accounts – as the same dict shape lookup('users') yields. Every consumer that provisions or references directory users must read this, so no two of them can disagree about who exists.

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[dict[str, Any]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.version module¶

Ansible lookup plugin: version

Reads pyproject.toml from the repository root, located by walking parents from this plugin file’s location.

param terms:: empty for the project’s own version, resolved as 1) [project].version (PEP 621) 2) [tool.poetry].version (Poetry fallback); or one distribution name, for the version its exact == pin in [project.dependencies] names.

class plugins.lookup.version.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms, variables=None, **kwargs)¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)

plugins.lookup.volume module¶

Lookup volume: read the canonical meta/volumes.yml registry.

Two call forms:

lookup('volume', application_id, name) — return the normalised dict for a single semantic entry. The returned dict always carries:

name the EFFECTIVE docker volume name
(entry’s name: if set, otherwise the YAML key)

semantic_name the YAML key from meta/volumes.yml

docker_name the explicit name: field (alias for name
during the dict-of-dicts migration), '' if absent

path the legacy path: field, '' if absent

type one of volume | bind | config | secret

tmpfs, defaults to volume

source for binds / configs / secrets, '' if absent

nfs NFS opt-in (bool or dict), None if absent

mounts the raw mounts list, [] if absent

targets {service: container path} built from mounts

target the one container path this entry mounts at, present
only when every mount agrees. An entry that lands at different paths per service omits the key rather than picking one; read targets[service] there.

target is what lets meta/volumes.yml stay the single source of a container path: a role defines its constant from the registry

PROMETHEUS_CONFIG_FILE: “{{ lookup(‘volume’, application_id, ‘config’).target }}”

rather than repeating the literal, which nothing keeps in step with the mount it describes.

lookup('volume', application_id) — return the WHOLE canonical registry as a dict keyed by semantic name. Useful for tasks that loop over every declared volume (e.g. NFS subdir pre-creation). Returns {} for roles with no meta/volumes.yml.

class plugins.lookup.volume.LookupModule(loader: _dataloader.DataLoader | None = None, templar: _template.Templar | None = None, **kwargs)¶

Bases: LookupBase

run(terms: list[Any] | None, variables: dict[str, Any] | None = None, **kwargs: Any) → list[dict[str, Any]]¶

"{{ lookup('url', 'https://toshio.fedorapeople.org/one.txt', validate_certs=True) }}"

would end up calling the lookup plugin named url’s run method like this::: run([’https://toshio.fedorapeople.org/one.txt’], variables=available_variables, validate_certs=True)

Errors encountered during execution should be returned by raising AnsibleError() with a message describing the error.

Any strings returned by this method that could ever contain non-ascii must be converted into python’s unicode type as the strings will be run through jinja2 which has this requirement. You can use:

from ansible.module_utils.common.text.converters import to_text
result_string = to_text(result_string)