Compose Ansible with Python: custom filter plugins (reusing range/MAC parsing), a check-mode-aware custom module, a dynamic inventory script, an end-to-end SoT→render→diff→deploy flow, and pytest/molecule/lint testing for CI.
Where Ansible ends, Python begins — and the two compose. This lab wires the Python you've already written (range parsing, MAC normalization, LLDP graphs from Ranges & Parsing and Graphs & Routing) into Ansible as custom modules, filter plugins, and a dynamic inventory, then runs a safe SoT→render→diff→deploy flow.
The mental split: filter plugins transform data inside templates; modules do work against a device and report changed/ok; inventory plugins/scripts tell Ansible which hosts exist. All three are just Python.
Drop Python functions into filter_plugins/ and call them in templates. Here we reuse the VLAN-range
and MAC helpers you already have.
# filter_plugins/net_filters.py
def expand_vlans(spec):
"""'10-12,20' -> [10, 11, 12, 20] (your ranges-parsing code)"""
out = []
for part in str(spec).split(","):
part = part.strip()
if "-" in part:
a, b = part.split("-")
out.extend(range(int(a), int(b) + 1))
elif part:
out.append(int(part))
return sorted(set(out))
def normalize_mac(mac):
"""Any MAC format -> aabb.ccdd.eeff"""
hexs = "".join(c for c in mac.lower() if c in "0123456789abcdef")
return ".".join(hexs[i:i + 4] for i in range(0, 12, 4))
class FilterModule(object):
def filters(self):
return {"expand_vlans": expand_vlans, "normalize_mac": normalize_mac}
# templates/vlans.j2 — now the SoT can say "10-12,20"
{% for vid in vlan_spec | expand_vlans %}
vlan {{ vid }}
{% endfor %}
Point: business logic lives in tested Python; the template stays trivial.
A module lives in library/, takes an argument_spec, honors check mode,
and returns changed. Skeleton every custom module follows:
# library/vlan_enforcer.py
from ansible.module_utils.basic import AnsibleModule
def run():
module = AnsibleModule(
argument_spec=dict(
want=dict(type="list", required=True), # desired VLAN ids
have=dict(type="list", required=True), # current VLAN ids
),
supports_check_mode=True, # enables --check
)
want = set(module.params["want"])
have = set(module.params["have"])
to_add, to_del = sorted(want - have), sorted(have - want)
changed = bool(to_add or to_del)
if module.check_mode or not changed:
module.exit_json(changed=changed, to_add=to_add, to_del=to_del)
# ... apply to_add / to_del here (netmiko/napalm/api) ...
module.exit_json(changed=changed, to_add=to_add, to_del=to_del)
if __name__ == "__main__":
run()
Point: compute the delta, respect check_mode, and only report changed=True when
something actually differs — that's idempotency.
Any executable that prints the right JSON for --list works as inventory.
#!/usr/bin/env python3
# inventory/sot_inventory.py (chmod +x)
import json, sys
# your SoT: could be YAML, a DB, NetBox, an LLDP graph, etc.
SOT = {
"edge": ["r1", "r2"],
"core": ["c1"],
}
def build():
inv = {"_meta": {"hostvars": {}}}
for group, hosts in SOT.items():
inv[group] = {"hosts": hosts}
for h in hosts:
inv["_meta"]["hostvars"][h] = {
"ansible_network_os": "cisco.ios.ios",
"ansible_connection": "ansible.netcommon.network_cli",
}
return inv
if __name__ == "__main__":
if "--list" in sys.argv:
print(json.dumps(build()))
elif "--host" in sys.argv:
print(json.dumps({})) # per-host vars come from _meta
ansible-inventory -i inventory/sot_inventory.py --graph
ansible-playbook -i inventory/sot_inventory.py site.yml --check --diff
vlan_spec: "10-12,20".expand_vlans filter.ansible-playbook site.yml --check --diff shows exactly what would
change; nothing is touched.--check; serial: 1 + a post-check
guard the blast radius.*_facts task confirms the VLANs exist; a second run reports
0 changed (proof of idempotency).# tests/test_filters.py
from filter_plugins.net_filters import expand_vlans, normalize_mac
def test_expand_vlans():
assert expand_vlans("10-12,20") == [10, 11, 12, 20]
def test_normalize_mac():
assert normalize_mac("AA:BB:CC:DD:EE:FF") == "aabb.ccdd.eeff"
command/uri? (idempotent, check-mode-aware, reusable work)--list / --host)--check? (supports_check_mode + return before applying)