mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
f75d3ee5f6
The new role-templates lookup read its paths with os.path, which the repo's ruff config rejects (PTH113, PTH119). autoformat is the first step of the quality gate, so those two lines stopped the gate before a single test ran -- the whole suite was unreachable over a filesystem call spelled the old way. Behaviour is unchanged: Path(...).is_file() and Path(...).name answer exactly what os.path.isfile and os.path.basename answered. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import glob
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from ansible.errors import AnsibleError
|
|
from ansible.plugins.lookup import LookupBase
|
|
|
|
from utils import PROJECT_ROOT
|
|
|
|
DEFAULTS: dict[str, Any] = {
|
|
"owner": "root",
|
|
"group": "root",
|
|
"mode": "0640",
|
|
"condition": True,
|
|
"notify": ["compose-up"],
|
|
}
|
|
|
|
|
|
def _glob_files(pattern: str) -> list[str]:
|
|
"""Paths matching *pattern*, files only, sorted.
|
|
|
|
Args:
|
|
pattern: an absolute glob pattern.
|
|
|
|
Returns:
|
|
Sorted absolute paths; directories are dropped, mirroring Ansible's
|
|
``fileglob``.
|
|
"""
|
|
return sorted(path for path in glob.glob(pattern) if Path(path).is_file())
|
|
|
|
|
|
def _strip_j2(name: str) -> str:
|
|
"""*name* without a trailing ``.j2``."""
|
|
return name.removesuffix(".j2")
|
|
|
|
|
|
def _expand_glob_entry(entry: dict, application_id: str) -> list[dict]:
|
|
"""One ``src_glob`` entry turned into one dict per matched template.
|
|
|
|
Args:
|
|
entry: the role_templates entry carrying ``src_glob`` and ``dest_dir``.
|
|
application_id: role whose ``templates/`` directory is globbed.
|
|
|
|
Returns:
|
|
One rendered-template spec per match, sorted by source path.
|
|
|
|
Raises:
|
|
AnsibleError: ``dest_dir`` is missing or not a string.
|
|
"""
|
|
dest_dir = entry.get("dest_dir")
|
|
if not isinstance(dest_dir, str) or not dest_dir:
|
|
raise AnsibleError(
|
|
"role_templates_expanded: an entry with 'src_glob' requires a "
|
|
f"non-empty string 'dest_dir', got {dest_dir!r}"
|
|
)
|
|
|
|
templates_dir = PROJECT_ROOT / "roles" / application_id / "templates"
|
|
out: list[dict] = []
|
|
for path in _glob_files(str(templates_dir / str(entry["src_glob"]))):
|
|
item = {key: entry.get(key, value) for key, value in DEFAULTS.items()}
|
|
item["src"] = path
|
|
item["dest"] = dest_dir.rstrip("/") + "/" + _strip_j2(Path(path).name)
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
class LookupModule(LookupBase):
|
|
def run(self, terms, variables: dict | None = None, **kwargs):
|
|
"""Expand ``role_templates`` into one spec per template to render.
|
|
|
|
Args:
|
|
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.
|
|
"""
|
|
if not terms or len(terms) != 2:
|
|
raise AnsibleError(
|
|
"role_templates_expanded: exactly two terms required "
|
|
"(application_id, role_templates)"
|
|
)
|
|
|
|
application_id = str(terms[0] or "").strip()
|
|
if not application_id:
|
|
raise AnsibleError("role_templates_expanded: application_id is empty")
|
|
|
|
entries = terms[1]
|
|
if entries is None:
|
|
return [[]]
|
|
if not isinstance(entries, list):
|
|
raise AnsibleError(
|
|
f"role_templates_expanded: role_templates must be a list, got {type(entries)}"
|
|
)
|
|
|
|
out: list[dict] = []
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
raise AnsibleError(
|
|
f"role_templates_expanded: every entry must be a dict, got {type(entry)}"
|
|
)
|
|
if "src_glob" in entry:
|
|
out.extend(_expand_glob_entry(entry, application_id))
|
|
continue
|
|
item = dict(entry)
|
|
for key, value in DEFAULTS.items():
|
|
item.setdefault(key, value)
|
|
out.append(item)
|
|
return [out]
|