82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
"""Ansible --list inventory generation from DB records."""
|
|
from __future__ import annotations
|
|
|
|
|
|
def generate_inventory(targets: list) -> dict:
|
|
"""Build an Ansible --list JSON inventory from a list of AnsibleTarget objects.
|
|
|
|
Accepts any objects with .name, .address, .ansible_vars (dict), .groups (list
|
|
of objects with .name and .ansible_vars). Uses duck typing so unit tests can
|
|
pass SimpleNamespace objects.
|
|
|
|
Var precedence (lowest → highest, matching Ansible default):
|
|
group vars (sorted alphabetically by group name) → target ansible_vars → ansible_host
|
|
"""
|
|
inv: dict = {"all": {"hosts": []}, "_meta": {"hostvars": {}}}
|
|
|
|
for target in targets:
|
|
inv["all"]["hosts"].append(target.name)
|
|
|
|
# Merge vars: group vars alphabetically, then host vars, then force ansible_host.
|
|
merged: dict = {}
|
|
for group in sorted(target.groups, key=lambda g: g.name):
|
|
merged.update(group.ansible_vars or {})
|
|
merged.update(target.ansible_vars or {})
|
|
merged["ansible_host"] = target.address
|
|
|
|
inv["_meta"]["hostvars"][target.name] = merged
|
|
|
|
for group in target.groups:
|
|
if group.name not in inv:
|
|
inv[group.name] = {
|
|
"hosts": [],
|
|
"vars": dict(group.ansible_vars or {}),
|
|
}
|
|
inv[group.name]["hosts"].append(target.name)
|
|
|
|
return inv
|
|
|
|
|
|
async def fetch_scope_targets(db, scope: str) -> list:
|
|
"""Query AnsibleTarget objects for a given inventory scope string.
|
|
|
|
Scopes:
|
|
steward:all — every target
|
|
steward:group:<uuid> — targets in that group
|
|
steward:target:<uuid> — single target by ID
|
|
|
|
Returns a list with groups pre-loaded (selectinload). Returns [] for
|
|
repo:* scopes (caller handles those via legacy path).
|
|
"""
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
|
|
|
|
if scope == "steward:all":
|
|
stmt = (
|
|
select(AnsibleTarget)
|
|
.options(selectinload(AnsibleTarget.groups))
|
|
.order_by(AnsibleTarget.name)
|
|
)
|
|
elif scope.startswith("steward:group:"):
|
|
group_id = scope[len("steward:group:"):]
|
|
stmt = (
|
|
select(AnsibleTarget)
|
|
.join(AnsibleTarget.groups)
|
|
.where(AnsibleGroup.id == group_id)
|
|
.options(selectinload(AnsibleTarget.groups))
|
|
.order_by(AnsibleTarget.name)
|
|
)
|
|
elif scope.startswith("steward:target:"):
|
|
target_id = scope[len("steward:target:"):]
|
|
stmt = (
|
|
select(AnsibleTarget)
|
|
.where(AnsibleTarget.id == target_id)
|
|
.options(selectinload(AnsibleTarget.groups))
|
|
)
|
|
else:
|
|
return []
|
|
|
|
result = await db.execute(stmt)
|
|
return list(result.scalars().all())
|