API Reference
Auto-generated reference documentation for the hier_config public API.
Constructor Functions
hier_config.get_hconfig(platform_or_driver, config_raw='')
Source code in hier_config/constructors.py
def get_hconfig(
platform_or_driver: Platform | HConfigDriverBase,
config_raw: Path | str = "",
) -> HConfig:
if isinstance(config_raw, Path):
config_raw = config_raw.read_text(encoding="utf8")
config = HConfig(_get_driver(platform_or_driver))
for rule in config.driver.rules.full_text_sub:
config_raw = sub(rule.search, rule.replace, config_raw)
_load_from_string_lines(config, config_raw)
for child in tuple(config.all_children()):
child.delete_sectional_exit()
for callback in config.driver.rules.post_load_callbacks:
callback(config)
return config
hier_config.get_hconfig_driver(platform)
Create base options on an OS level.
Source code in hier_config/constructors.py
def get_hconfig_driver(platform: Platform) -> HConfigDriverBase:
"""Create base options on an OS level."""
platform_drivers: dict[Platform, type[HConfigDriverBase]] = {
Platform.ARISTA_EOS: HConfigDriverAristaEOS,
Platform.CISCO_IOS: HConfigDriverCiscoIOS,
Platform.CISCO_NXOS: HConfigDriverCiscoNXOS,
Platform.CISCO_XR: HConfigDriverCiscoIOSXR,
Platform.FORTINET_FORTIOS: HConfigDriverFortinetFortiOS,
Platform.GENERIC: HConfigDriverGeneric,
Platform.HP_PROCURVE: HConfigDriverHPProcurve,
Platform.HP_COMWARE5: HConfigDriverHPComware5,
Platform.HUAWEI_VRP: HConfigDriverHuaweiVrp,
Platform.JUNIPER_JUNOS: HConfigDriverJuniperJUNOS,
Platform.NOKIA_SRL: HConfigDriverNokiaSRL,
Platform.VYOS: HConfigDriverVYOS,
}
driver_cls = platform_drivers.get(platform)
if driver_cls is None:
message = f"Unsupported platform: {platform}"
raise ValueError(message)
return driver_cls()
Core Classes
hier_config.HConfig
Bases: HConfigBase
A class for representing and comparing Cisco like configurations in a hierarchical tree data structure.
Source code in hier_config/root.py
class HConfig(HConfigBase): # ruff:ignore[too-many-public-methods]
"""A class for representing and comparing Cisco like configurations in a
hierarchical tree data structure.
"""
__slots__ = ("_driver",)
def __init__(self, driver: HConfigDriverBase) -> None:
super().__init__()
self._driver = driver
def __str__(self) -> str:
return "\n".join(str(c) for c in sorted(self.children))
def __repr__(self) -> str:
return f"HConfig(driver={self.driver.__class__.__name__}, lines={self.dump_simple()})"
def __hash__(self) -> int:
return hash(*self.children)
def __eq__(self, other: object) -> bool:
if not isinstance(other, HConfig):
return NotImplemented
return self.children == other.children
@property
def driver(self) -> HConfigDriverBase:
return self._driver
@property
def real_indent_level(self) -> int:
return -1
@property
def parent(self) -> HConfig:
return self
@property
def root(self) -> HConfig:
"""The HConfig object at the base of the tree."""
return self
@property
def is_leaf(self) -> bool:
"""True if there are no children and is not an instance of HConfig."""
return False
@property
def is_branch(self) -> bool:
"""True if there are children or is an instance of HConfig."""
return True
def instantiate_child(self, text: str) -> HConfigChild:
return HConfigChild(self, text)
@property
def tags(self) -> frozenset[str]:
"""Recursive access to tags on all leaf nodes."""
found_tags: set[str] = set()
for child in self.children:
found_tags.update(child.tags)
return frozenset(found_tags)
@tags.setter
def tags(self, value: frozenset[str]) -> None:
"""Recursive access to tags on all leaf nodes."""
for child in self.children:
child.tags = value
def merge(self, other: HConfig | Iterable[HConfig]) -> HConfig:
"""Merges other HConfig objects into this one."""
other_configs = (other,) if isinstance(other, HConfig) else other
for other_config in other_configs:
for child in other_config.children:
self.add_deep_copy_of(child, merged=True)
return self
def add_children_deep(self, lines: Iterable[str]) -> HConfigChild:
"""Add child instances of HConfigChild deeply."""
base: HConfig | HConfigChild = self
for line in lines:
base = base.add_child(line, return_if_present=True)
if isinstance(base, HConfig):
message = "base was an HConfig object for some reason."
raise TypeError(message)
return base
def lineage(self) -> Iterator[HConfigChild]: # ruff:ignore[no-self-use]
"""Yields the lineage of parent objects, up to but excluding the root."""
yield from ()
def lines(self, *, sectional_exiting: bool = False) -> Iterable[str]:
for child in sorted(self.children):
yield from child.lines(sectional_exiting=sectional_exiting)
def dump_simple(self, *, sectional_exiting: bool = False) -> tuple[str, ...]:
return tuple(self.lines(sectional_exiting=sectional_exiting))
def dump(self) -> Dump:
"""Dump loaded HConfig data."""
return Dump(
lines=tuple(
DumpLine(
depth=c.depth(),
text=c.text,
tags=frozenset(c.tags),
comments=frozenset(c.comments),
new_in_config=c.new_in_config,
)
for c in self.all_children_sorted()
),
)
def depth(self) -> int: # ruff:ignore[no-self-use]
"""The distance to the root HConfig object i.e. indent level."""
return 0
def difference(self, target: HConfig) -> HConfig:
"""Creates a new HConfig object with the config from self that is not in target."""
return self._difference(target, HConfig(self.driver))
def config_to_get_to(
self,
target: HConfig,
delta: HConfig | None = None,
) -> HConfig:
"""Figures out what commands need to be executed to transition from self to target.
self is the source data structure(i.e. the running_config),
target is the destination(i.e. generated_config).
"""
if delta is None:
delta = HConfig(self.driver)
return self._config_to_get_to(target, delta)
def add_ancestor_copy_of(
self,
parent_to_add: HConfigChild,
) -> HConfig | HConfigChild:
"""Add a copy of the ancestry of parent_to_add to self
and return the deepest child which is equivalent to parent_to_add.
"""
base: HConfig | HConfigChild = self
for parent in parent_to_add.lineage():
base = base.add_shallow_copy_of(parent)
return base
def set_order_weight(self) -> HConfig:
"""Sets self.order integer on all children."""
for child in self.all_children():
for rule in self.driver.rules.ordering:
if child.is_lineage_match(rule.match_rules):
child.order_weight = rule.weight
return self
def future(
self,
config: HConfig,
*,
prune_empty_branches: bool = False,
) -> HConfig:
"""EXPERIMENTAL - predict the future config after config is applied to self.
The quality of this method's output will in part depend on how well
the OS options are tuned. Ensuring that idempotency rules are accurate is
especially important.
With `prune_empty_branches`, sections that the change emptied out are
removed, matching devices that prune empty stanzas on commit; sections
that were already empty are kept.
"""
future_config = HConfig(self.driver)
self._future(config, future_config)
if prune_empty_branches:
self._prune_emptied_branches(future_config)
return future_config
def with_tags(self, tags: Iterable[str]) -> HConfig:
"""Returns a new instance recursively containing children that only have a subset of tags."""
return self._with_tags(frozenset(tags), HConfig(self.driver))
def all_children_sorted_by_tags(
self,
include_tags: Iterable[str],
exclude_tags: Iterable[str],
) -> Iterator[HConfigChild]:
"""Yield all children recursively that match include/exclude tags."""
for child in sorted(self.children):
yield from child.all_children_sorted_by_tags(include_tags, exclude_tags)
def deep_copy(self) -> HConfig:
"""Return a copy of this object."""
new_instance = HConfig(self.driver)
for child in self.children:
new_instance.add_deep_copy_of(child)
return new_instance
def unused_objects(self) -> Iterator[HConfigChild]:
"""Yield top-level children that are defined objects with no references.
Uses ``self.driver.rules.unused_objects`` to identify object definitions,
extract their names, and search for references across the config tree.
Objects with zero references are yielded.
"""
from re import search as _re_search # ruff:ignore[import-outside-top-level]
for rule in self.driver.rules.unused_objects:
seen_names: set[str] = set()
for definition in self.get_children_deep(rule.match_rules):
match = _re_search(rule.name_re, definition.text)
if not match:
continue
name = match.group("name")
if name in seen_names:
continue
seen_names.add(name)
if not self._is_object_referenced(name, rule.reference_locations):
yield definition
def _is_object_referenced(
self,
name: str,
reference_locations: tuple[ReferenceLocation, ...],
) -> bool:
"""Return True if *name* is found in any reference location."""
from re import escape as _re_escape # ruff:ignore[import-outside-top-level]
from re import search as _re_search # ruff:ignore[import-outside-top-level]
for ref_location in reference_locations:
pattern = ref_location.reference_re.format(name=_re_escape(name))
for section in self.get_children_deep(ref_location.match_rules):
if any(_re_search(pattern, d.text) for d in section.all_children()):
return True
return False
def _is_duplicate_child_allowed(self) -> bool: # ruff:ignore[no-self-use]
"""Determine if duplicate(identical text) children are allowed under the parent."""
return False
is_branch
property
True if there are children or is an instance of HConfig.
is_leaf
property
True if there are no children and is not an instance of HConfig.
root
property
The HConfig object at the base of the tree.
tags
property
writable
Recursive access to tags on all leaf nodes.
add_ancestor_copy_of(parent_to_add)
Add a copy of the ancestry of parent_to_add to self and return the deepest child which is equivalent to parent_to_add.
Source code in hier_config/root.py
def add_ancestor_copy_of(
self,
parent_to_add: HConfigChild,
) -> HConfig | HConfigChild:
"""Add a copy of the ancestry of parent_to_add to self
and return the deepest child which is equivalent to parent_to_add.
"""
base: HConfig | HConfigChild = self
for parent in parent_to_add.lineage():
base = base.add_shallow_copy_of(parent)
return base
add_children_deep(lines)
Add child instances of HConfigChild deeply.
Source code in hier_config/root.py
def add_children_deep(self, lines: Iterable[str]) -> HConfigChild:
"""Add child instances of HConfigChild deeply."""
base: HConfig | HConfigChild = self
for line in lines:
base = base.add_child(line, return_if_present=True)
if isinstance(base, HConfig):
message = "base was an HConfig object for some reason."
raise TypeError(message)
return base
all_children_sorted_by_tags(include_tags, exclude_tags)
Yield all children recursively that match include/exclude tags.
Source code in hier_config/root.py
def all_children_sorted_by_tags(
self,
include_tags: Iterable[str],
exclude_tags: Iterable[str],
) -> Iterator[HConfigChild]:
"""Yield all children recursively that match include/exclude tags."""
for child in sorted(self.children):
yield from child.all_children_sorted_by_tags(include_tags, exclude_tags)
config_to_get_to(target, delta=None)
Figures out what commands need to be executed to transition from self to target. self is the source data structure(i.e. the running_config), target is the destination(i.e. generated_config).
Source code in hier_config/root.py
def config_to_get_to(
self,
target: HConfig,
delta: HConfig | None = None,
) -> HConfig:
"""Figures out what commands need to be executed to transition from self to target.
self is the source data structure(i.e. the running_config),
target is the destination(i.e. generated_config).
"""
if delta is None:
delta = HConfig(self.driver)
return self._config_to_get_to(target, delta)
deep_copy()
Return a copy of this object.
Source code in hier_config/root.py
def deep_copy(self) -> HConfig:
"""Return a copy of this object."""
new_instance = HConfig(self.driver)
for child in self.children:
new_instance.add_deep_copy_of(child)
return new_instance
depth()
The distance to the root HConfig object i.e. indent level.
Source code in hier_config/root.py
def depth(self) -> int: # ruff:ignore[no-self-use]
"""The distance to the root HConfig object i.e. indent level."""
return 0
difference(target)
Creates a new HConfig object with the config from self that is not in target.
Source code in hier_config/root.py
def difference(self, target: HConfig) -> HConfig:
"""Creates a new HConfig object with the config from self that is not in target."""
return self._difference(target, HConfig(self.driver))
dump()
Dump loaded HConfig data.
Source code in hier_config/root.py
def dump(self) -> Dump:
"""Dump loaded HConfig data."""
return Dump(
lines=tuple(
DumpLine(
depth=c.depth(),
text=c.text,
tags=frozenset(c.tags),
comments=frozenset(c.comments),
new_in_config=c.new_in_config,
)
for c in self.all_children_sorted()
),
)
future(config, *, prune_empty_branches=False)
EXPERIMENTAL - predict the future config after config is applied to self.
The quality of this method's output will in part depend on how well the OS options are tuned. Ensuring that idempotency rules are accurate is especially important.
With prune_empty_branches, sections that the change emptied out are
removed, matching devices that prune empty stanzas on commit; sections
that were already empty are kept.
Source code in hier_config/root.py
def future(
self,
config: HConfig,
*,
prune_empty_branches: bool = False,
) -> HConfig:
"""EXPERIMENTAL - predict the future config after config is applied to self.
The quality of this method's output will in part depend on how well
the OS options are tuned. Ensuring that idempotency rules are accurate is
especially important.
With `prune_empty_branches`, sections that the change emptied out are
removed, matching devices that prune empty stanzas on commit; sections
that were already empty are kept.
"""
future_config = HConfig(self.driver)
self._future(config, future_config)
if prune_empty_branches:
self._prune_emptied_branches(future_config)
return future_config
lineage()
Yields the lineage of parent objects, up to but excluding the root.
Source code in hier_config/root.py
def lineage(self) -> Iterator[HConfigChild]: # ruff:ignore[no-self-use]
"""Yields the lineage of parent objects, up to but excluding the root."""
yield from ()
merge(other)
Merges other HConfig objects into this one.
Source code in hier_config/root.py
def merge(self, other: HConfig | Iterable[HConfig]) -> HConfig:
"""Merges other HConfig objects into this one."""
other_configs = (other,) if isinstance(other, HConfig) else other
for other_config in other_configs:
for child in other_config.children:
self.add_deep_copy_of(child, merged=True)
return self
set_order_weight()
Sets self.order integer on all children.
Source code in hier_config/root.py
def set_order_weight(self) -> HConfig:
"""Sets self.order integer on all children."""
for child in self.all_children():
for rule in self.driver.rules.ordering:
if child.is_lineage_match(rule.match_rules):
child.order_weight = rule.weight
return self
unused_objects()
Yield top-level children that are defined objects with no references.
Uses self.driver.rules.unused_objects to identify object definitions,
extract their names, and search for references across the config tree.
Objects with zero references are yielded.
Source code in hier_config/root.py
def unused_objects(self) -> Iterator[HConfigChild]:
"""Yield top-level children that are defined objects with no references.
Uses ``self.driver.rules.unused_objects`` to identify object definitions,
extract their names, and search for references across the config tree.
Objects with zero references are yielded.
"""
from re import search as _re_search # ruff:ignore[import-outside-top-level]
for rule in self.driver.rules.unused_objects:
seen_names: set[str] = set()
for definition in self.get_children_deep(rule.match_rules):
match = _re_search(rule.name_re, definition.text)
if not match:
continue
name = match.group("name")
if name in seen_names:
continue
seen_names.add(name)
if not self._is_object_referenced(name, rule.reference_locations):
yield definition
with_tags(tags)
Returns a new instance recursively containing children that only have a subset of tags.
Source code in hier_config/root.py
def with_tags(self, tags: Iterable[str]) -> HConfig:
"""Returns a new instance recursively containing children that only have a subset of tags."""
return self._with_tags(frozenset(tags), HConfig(self.driver))
hier_config.HConfigChild
Bases: HConfigBase
A single node in the hierarchical configuration tree.
Each HConfigChild holds one configuration line (text), an ordered
collection of its own children, optional tags/comments, and a reference
back to its parent. The tree is rooted at an HConfig instance; every
other node is an HConfigChild.
Source code in hier_config/child.py
class HConfigChild( # ruff:ignore[too-many-public-methods] pylint: disable=too-many-instance-attributes
HConfigBase,
):
"""A single node in the hierarchical configuration tree.
Each `HConfigChild` holds one configuration line (`text`), an ordered
collection of its own children, optional tags/comments, and a reference
back to its parent. The tree is rooted at an `HConfig` instance; every
other node is an `HConfigChild`.
"""
__slots__ = (
"_tags",
"_text",
"comments",
"facts",
"instances",
"new_in_config",
"order_weight",
"parent",
"real_indent_level",
)
def __init__(self, parent: HConfig | HConfigChild, text: str) -> None:
super().__init__()
self.parent = parent
self._text: str = text.strip()
self.real_indent_level: int
# 0 is the default. Positive weights sink while negative weights rise.
self.order_weight: int = 0
self._tags: set[str] = set()
self.comments: set[str] = set()
self.new_in_config: bool = False
self.instances: list[Instance] = []
# To store externally inserted facts
self.facts: dict[Any, Any] = {}
def __str__(self) -> str:
return "\n".join(self.lines(sectional_exiting=True))
def __repr__(self) -> str:
return f"HConfigChild(HConfig{'' if self.parent is self.root else 'Child'}, {self.text})"
def __lt__(self, other: HConfigChild) -> bool:
return self.order_weight < other.order_weight
def __hash__(self) -> int:
return hash(
(
self.text,
self.tags,
*self.children,
),
)
def __eq__(self, other: object) -> bool:
if not isinstance(other, HConfigChild):
return NotImplemented
# We are intentionally not including the
# comments, facts, instances, new_in_config, order_weight attributes.
if self.text != other.text or self.tags != other.tags:
return False
return self.children == other.children
def __ne__(self, other: object) -> bool:
return not self.__eq__(other)
@property
def driver(self) -> HConfigDriverBase:
return self.root.driver
@property
def text(self) -> str:
return self._text
@text.setter
def text(self, value: str) -> None:
"""Used for when self.text is changed after the object
is instantiated to rebuild the children dictionary.
"""
self._text = value.strip()
self.parent.children.rebuild_mapping()
@property
def text_without_negation(self) -> str:
return self.text.removeprefix(self.driver.negation_prefix)
@property
def root(self) -> HConfig:
"""The HConfig object at the base of the tree."""
return self.parent.root
def lines(self, *, sectional_exiting: bool = False) -> Iterable[str]:
yield self.cisco_style_text()
for child in sorted(self.children):
yield from child.lines(sectional_exiting=sectional_exiting)
if sectional_exiting and (exit_text := self.sectional_exit):
depth = (
self.depth() - 1
if self.sectional_exit_text_parent_level
else self.depth()
)
yield " " * self.driver.rules.indentation * depth + exit_text
@property
def sectional_exit(self) -> str | None:
return self.driver.sectional_exit(self)
@property
def sectional_exit_text_parent_level(self) -> bool:
for rule in self.driver.rules.sectional_exiting:
if self.is_lineage_match(rule.match_rules):
return rule.exit_text_parent_level
return False
def delete_sectional_exit(self) -> None:
try:
potential_exit = self.children[-1]
except IndexError:
return
if (exit_text := self.sectional_exit) and exit_text == potential_exit.text:
potential_exit.delete()
def depth(self) -> int:
"""The distance to the root HConfig object i.e. indent level."""
return self.parent.depth() + 1
def move(self, new_parent: HConfig | HConfigChild) -> None:
"""Move one HConfigChild object to different HConfig parent object.
.. code:: python
hier1 = config_for_platform(host.platform)
interface1 = hier1.add_child('interface Vlan2')
interface1.add_child('ip address 10.0.0.1 255.255.255.252')
hier2 = Hconfig(host)
interface1.move(hier2)
:param new_parent: HConfigChild object -> type list
"""
new_parent.children.append(self)
self.delete()
def lineage(self) -> Iterator[HConfigChild]:
"""Yields the lineage of parent objects up to, but excluding, the root."""
yield from self.parent.lineage()
yield self
def path(self) -> Iterator[str]:
"""Yields the text attribute of child objects up to, but excluding, the root."""
for child in self.lineage():
yield child.text
def cisco_style_text(
self,
style: TextStyle = "without_comments",
tag: str | None = None,
) -> str:
"""Return a Cisco style formated line i.e. indentation_level + text ! comments."""
comments: list[str] = []
if style == "without_comments":
pass
elif style == "merged":
# count the number of instances that have the tag
instance_count = 0
instance_comments: set[str] = set()
for instance in self.instances:
if tag is None or tag in instance.tags:
instance_count += 1
instance_comments.update(instance.comments)
# should the word 'instance' be plural?
word = "instance" if instance_count == 1 else "instances"
comments.append(f"{instance_count} {word}")
comments.extend(instance_comments)
elif style == "with_comments":
comments.extend(self.comments)
comments_str = f" !{', '.join(sorted(comments))}" if comments else ""
return f"{self.indentation}{self.text}{comments_str}"
@property
def indentation(self) -> str:
return " " * self.driver.rules.indentation * (self.depth() - 1)
def delete(self) -> None:
"""Delete the current object from its parent."""
self.parent.children.delete(self)
def tags_add(self, tag: str | Iterable[str]) -> None:
"""Add a tag to self._tags on all leaf nodes."""
if self.is_branch:
for child in self.children:
child.tags_add(tag)
elif isinstance(tag, str):
self._tags.add(tag)
else:
self._tags.update(tag)
def tags_remove(self, tag: str | Iterable[str]) -> None:
"""Remove a tag from self._tags on all leaf nodes."""
if self.is_branch:
for child in self.children:
child.tags_remove(tag)
elif isinstance(tag, str):
self._tags.remove(tag)
else:
self._tags.difference_update(tag)
def negate(self) -> HConfigChild:
"""Negate self.text using driver-specific negation rules.
Negation is resolved in the following priority order:
1. ``negate_with`` rule — replaces ``self.text`` with a custom
negation string defined in the driver (e.g. ``no ip route``).
2. ``negation_default_when`` rule — rewrites the command to its
``default`` form (e.g. ``no shutdown`` → ``default shutdown``).
3. ``negation_sub`` rule — applies a regex substitution to the
negated text (e.g. truncating after a specific token).
4. ``swap_negation`` — toggles the negation prefix/declaration
prefix (e.g. ``shutdown`` ↔ ``no shutdown``).
Returns self so that callers can chain further operations.
"""
if negate_with := self.driver.negate_with(self):
self.text = negate_with
return self
if self.use_default_for_negation(self):
return self._default()
for rule in self.driver.rules.negation_sub:
if self.is_lineage_match(rule.match_rules):
self.text = sub(
rule.search,
rule.replace,
f"{self.driver.negation_prefix}{self.text}",
)
return self
return self.driver.swap_negation(self)
def use_default_for_negation(self, config: HConfigChild) -> bool:
return any(
config.is_lineage_match(rule.match_rules)
for rule in self.driver.rules.negation_default_when
)
@property
def is_leaf(self) -> bool:
"""True if there are no children and is not an instance of HConfig."""
return not self.is_branch
@property
def is_branch(self) -> bool:
"""True if there are children or is an instance of HConfig."""
return bool(self.children)
@property
def tags(self) -> frozenset[str]:
"""Recursive access to tags on all leaf nodes."""
if self.is_branch:
found_tags: set[str] = set()
for child in self.children:
found_tags.update(child.tags)
return frozenset(found_tags)
return frozenset(self._tags)
@tags.setter
def tags(self, value: frozenset[str]) -> None:
"""Recursive access to tags on all leaf nodes."""
if self.is_branch:
for child in self.children:
child.tags = value
else:
self._tags = set(value)
def is_idempotent_command(self, other_children: Iterable[HConfigChild]) -> bool:
"""Determine if self.text is an idempotent change."""
# Avoid list commands from matching as idempotent
for rule in self.driver.rules.idempotent_commands_avoid:
if self.is_lineage_match(rule.match_rules):
return False
# Idempotent command identification
return bool(self.driver.idempotent_for(self, other_children))
def use_sectional_overwrite_without_negation(self) -> bool:
"""Check self's text to see if negation should be handled by
overwriting the section without first negating it.
"""
return any(
self.is_lineage_match(rule.match_rules)
for rule in self.driver.rules.sectional_overwrite_no_negate
)
def use_sectional_overwrite(self) -> bool:
"""Determines if self.text matches a sectional overwrite rule."""
return any(
self.is_lineage_match(rule.match_rules)
for rule in self.driver.rules.sectional_overwrite
)
def overwrite_with(
self,
target: HConfigChild,
delta: HConfig | HConfigChild,
*,
negate: bool = True,
) -> None:
"""Overwrite self's section in delta with a deep copy of target.
When the children of self and target differ, this method mutates
``delta`` in-place: the existing entry for ``self.text`` is negated
(if ``negate=True``) or simply deleted (if ``negate=False``), and a
fresh deep copy of ``target`` is appended. A ``"re-create section"``
comment is attached to the new entry, and a ``"dropping section"``
comment is added to the negated entry when applicable.
Used by :meth:`_config_to_get_to_right` when a sectional-overwrite
rule is active for ``self.text``.
"""
if self.children != target.children:
if negate:
if negated := delta.children.get(self.text):
negated.negate()
else:
negated = delta.add_child(
self.text, check_if_present=False
).negate()
negated.comments.add("dropping section")
else:
delta.children.delete(self.text)
if self.children:
new_item = delta.add_deep_copy_of(target)
new_item.comments.add("re-create section")
def line_inclusion_test(
self,
include_tags: Iterable[str],
exclude_tags: Iterable[str],
) -> bool:
"""Given the line_tags, include_tags, and exclude_tags,
determine if the line should be included.
"""
include_line = False
if include_tags:
include_line = bool(self.tags.intersection(include_tags))
if exclude_tags and (include_line or not include_tags):
return not bool(self.tags.intersection(exclude_tags))
return include_line
@property
def instance(self) -> Instance:
return Instance(
id=id(self.root),
comments=frozenset(self.comments),
tags=frozenset(self.tags),
)
def all_children_sorted_by_tags(
self,
include_tags: Iterable[str],
exclude_tags: Iterable[str],
) -> Iterator[HConfigChild]:
"""Yield all children recursively that match include/exclude tags."""
if self.is_leaf:
if self.line_inclusion_test(include_tags, exclude_tags):
yield self
else:
self_iter = iter((self,))
for child in sorted(self.children):
included_children = child.all_children_sorted_by_tags(
include_tags,
exclude_tags,
)
if peek := next(included_children, None):
yield from chain(self_iter, (peek,), included_children)
def is_lineage_match(self, rules: tuple[MatchRule, ...]) -> bool:
"""A generic test against a lineage of HConfigChild objects."""
lineage = tuple(self.lineage())
return len(rules) == len(lineage) and all(
child.is_match(
equals=rule.equals,
startswith=rule.startswith,
endswith=rule.endswith,
contains=rule.contains,
re_search=rule.re_search,
)
for (child, rule) in zip(reversed(lineage), reversed(rules), strict=True)
)
def is_match( # ruff:ignore[too-many-return-statements]
self,
*,
equals: str | SetLikeOfStr | None = None,
startswith: str | tuple[str, ...] | None = None,
endswith: str | tuple[str, ...] | None = None,
contains: str | tuple[str, ...] | None = None,
re_search: str | None = None,
) -> bool:
"""Return True if ``self.text`` satisfies all supplied criteria.
All arguments are optional. When *all* arguments are ``None`` the
method returns ``True`` (matches everything). When multiple arguments
are provided, **all** must match.
Args:
equals: Exact string match, or a frozenset of acceptable values.
startswith: ``str.startswith`` prefix (str or tuple of strs).
endswith: ``str.endswith`` suffix (str or tuple of strs).
contains: Substring(s) that must appear in ``self.text``.
re_search: Regular expression applied via :func:`re.search`.
Returns:
``True`` if every non-``None`` criterion is satisfied.
"""
# Equals filter
if isinstance(equals, str):
if self.text != equals:
return False
elif ( # pylint: disable=confusing-consecutive-elif
isinstance(equals, frozenset) and self.text not in equals
):
return False
# Startswith filter
if isinstance(startswith, (str, tuple)) and not self.text.startswith(
startswith
):
return False
# Regex filter
if isinstance(re_search, str) and not search(re_search, self.text):
return False
# The below filters are less commonly used
# Endswith filter
if isinstance(endswith, (str, tuple)) and not self.text.endswith(endswith):
return False
# Contains filter
if isinstance(contains, str):
if contains not in self.text:
return False
elif isinstance( # pylint: disable=confusing-consecutive-elif
contains,
tuple,
) and not any(c in self.text for c in contains):
return False
return True
def add_children_deep(self, lines: Iterable[str]) -> HConfigChild:
"""Add child instances of HConfigChild deeply."""
base = self
for line in lines:
base = base.add_child(line)
return base
def _default(self) -> HConfigChild:
"""Default self.text."""
self.text = f"default {self.text_without_negation}"
return self
def instantiate_child(self, text: str) -> HConfigChild:
return HConfigChild(self, text)
def _is_duplicate_child_allowed(self) -> bool:
"""Determine if duplicate(identical text) children are allowed under the parent."""
return any(
self.is_lineage_match(rule.match_rules)
for rule in self.driver.rules.parent_allows_duplicate_child
)
is_branch
property
True if there are children or is an instance of HConfig.
is_leaf
property
True if there are no children and is not an instance of HConfig.
root
property
The HConfig object at the base of the tree.
tags
property
writable
Recursive access to tags on all leaf nodes.
add_children_deep(lines)
Add child instances of HConfigChild deeply.
Source code in hier_config/child.py
def add_children_deep(self, lines: Iterable[str]) -> HConfigChild:
"""Add child instances of HConfigChild deeply."""
base = self
for line in lines:
base = base.add_child(line)
return base
all_children_sorted_by_tags(include_tags, exclude_tags)
Yield all children recursively that match include/exclude tags.
Source code in hier_config/child.py
def all_children_sorted_by_tags(
self,
include_tags: Iterable[str],
exclude_tags: Iterable[str],
) -> Iterator[HConfigChild]:
"""Yield all children recursively that match include/exclude tags."""
if self.is_leaf:
if self.line_inclusion_test(include_tags, exclude_tags):
yield self
else:
self_iter = iter((self,))
for child in sorted(self.children):
included_children = child.all_children_sorted_by_tags(
include_tags,
exclude_tags,
)
if peek := next(included_children, None):
yield from chain(self_iter, (peek,), included_children)
cisco_style_text(style='without_comments', tag=None)
Return a Cisco style formated line i.e. indentation_level + text ! comments.
Source code in hier_config/child.py
def cisco_style_text(
self,
style: TextStyle = "without_comments",
tag: str | None = None,
) -> str:
"""Return a Cisco style formated line i.e. indentation_level + text ! comments."""
comments: list[str] = []
if style == "without_comments":
pass
elif style == "merged":
# count the number of instances that have the tag
instance_count = 0
instance_comments: set[str] = set()
for instance in self.instances:
if tag is None or tag in instance.tags:
instance_count += 1
instance_comments.update(instance.comments)
# should the word 'instance' be plural?
word = "instance" if instance_count == 1 else "instances"
comments.append(f"{instance_count} {word}")
comments.extend(instance_comments)
elif style == "with_comments":
comments.extend(self.comments)
comments_str = f" !{', '.join(sorted(comments))}" if comments else ""
return f"{self.indentation}{self.text}{comments_str}"
delete()
Delete the current object from its parent.
Source code in hier_config/child.py
def delete(self) -> None:
"""Delete the current object from its parent."""
self.parent.children.delete(self)
depth()
The distance to the root HConfig object i.e. indent level.
Source code in hier_config/child.py
def depth(self) -> int:
"""The distance to the root HConfig object i.e. indent level."""
return self.parent.depth() + 1
is_idempotent_command(other_children)
Determine if self.text is an idempotent change.
Source code in hier_config/child.py
def is_idempotent_command(self, other_children: Iterable[HConfigChild]) -> bool:
"""Determine if self.text is an idempotent change."""
# Avoid list commands from matching as idempotent
for rule in self.driver.rules.idempotent_commands_avoid:
if self.is_lineage_match(rule.match_rules):
return False
# Idempotent command identification
return bool(self.driver.idempotent_for(self, other_children))
is_lineage_match(rules)
A generic test against a lineage of HConfigChild objects.
Source code in hier_config/child.py
def is_lineage_match(self, rules: tuple[MatchRule, ...]) -> bool:
"""A generic test against a lineage of HConfigChild objects."""
lineage = tuple(self.lineage())
return len(rules) == len(lineage) and all(
child.is_match(
equals=rule.equals,
startswith=rule.startswith,
endswith=rule.endswith,
contains=rule.contains,
re_search=rule.re_search,
)
for (child, rule) in zip(reversed(lineage), reversed(rules), strict=True)
)
is_match(*, equals=None, startswith=None, endswith=None, contains=None, re_search=None)
Return True if self.text satisfies all supplied criteria.
All arguments are optional. When all arguments are None the
method returns True (matches everything). When multiple arguments
are provided, all must match.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Source code in hier_config/child.py
def is_match( # ruff:ignore[too-many-return-statements]
self,
*,
equals: str | SetLikeOfStr | None = None,
startswith: str | tuple[str, ...] | None = None,
endswith: str | tuple[str, ...] | None = None,
contains: str | tuple[str, ...] | None = None,
re_search: str | None = None,
) -> bool:
"""Return True if ``self.text`` satisfies all supplied criteria.
All arguments are optional. When *all* arguments are ``None`` the
method returns ``True`` (matches everything). When multiple arguments
are provided, **all** must match.
Args:
equals: Exact string match, or a frozenset of acceptable values.
startswith: ``str.startswith`` prefix (str or tuple of strs).
endswith: ``str.endswith`` suffix (str or tuple of strs).
contains: Substring(s) that must appear in ``self.text``.
re_search: Regular expression applied via :func:`re.search`.
Returns:
``True`` if every non-``None`` criterion is satisfied.
"""
# Equals filter
if isinstance(equals, str):
if self.text != equals:
return False
elif ( # pylint: disable=confusing-consecutive-elif
isinstance(equals, frozenset) and self.text not in equals
):
return False
# Startswith filter
if isinstance(startswith, (str, tuple)) and not self.text.startswith(
startswith
):
return False
# Regex filter
if isinstance(re_search, str) and not search(re_search, self.text):
return False
# The below filters are less commonly used
# Endswith filter
if isinstance(endswith, (str, tuple)) and not self.text.endswith(endswith):
return False
# Contains filter
if isinstance(contains, str):
if contains not in self.text:
return False
elif isinstance( # pylint: disable=confusing-consecutive-elif
contains,
tuple,
) and not any(c in self.text for c in contains):
return False
return True
line_inclusion_test(include_tags, exclude_tags)
Given the line_tags, include_tags, and exclude_tags, determine if the line should be included.
Source code in hier_config/child.py
def line_inclusion_test(
self,
include_tags: Iterable[str],
exclude_tags: Iterable[str],
) -> bool:
"""Given the line_tags, include_tags, and exclude_tags,
determine if the line should be included.
"""
include_line = False
if include_tags:
include_line = bool(self.tags.intersection(include_tags))
if exclude_tags and (include_line or not include_tags):
return not bool(self.tags.intersection(exclude_tags))
return include_line
lineage()
Yields the lineage of parent objects up to, but excluding, the root.
Source code in hier_config/child.py
def lineage(self) -> Iterator[HConfigChild]:
"""Yields the lineage of parent objects up to, but excluding, the root."""
yield from self.parent.lineage()
yield self
move(new_parent)
Move one HConfigChild object to different HConfig parent object.
.. code:: python
hier1 = config_for_platform(host.platform)
interface1 = hier1.add_child('interface Vlan2')
interface1.add_child('ip address 10.0.0.1 255.255.255.252')
hier2 = Hconfig(host)
interface1.move(hier2)
:param new_parent: HConfigChild object -> type list
Source code in hier_config/child.py
def move(self, new_parent: HConfig | HConfigChild) -> None:
"""Move one HConfigChild object to different HConfig parent object.
.. code:: python
hier1 = config_for_platform(host.platform)
interface1 = hier1.add_child('interface Vlan2')
interface1.add_child('ip address 10.0.0.1 255.255.255.252')
hier2 = Hconfig(host)
interface1.move(hier2)
:param new_parent: HConfigChild object -> type list
"""
new_parent.children.append(self)
self.delete()
negate()
Negate self.text using driver-specific negation rules.
Negation is resolved in the following priority order:
negate_withrule — replacesself.textwith a custom negation string defined in the driver (e.g.no ip route).negation_default_whenrule — rewrites the command to itsdefaultform (e.g.no shutdown→default shutdown).negation_subrule — applies a regex substitution to the negated text (e.g. truncating after a specific token).swap_negation— toggles the negation prefix/declaration prefix (e.g.shutdown↔no shutdown).
Returns self so that callers can chain further operations.
Source code in hier_config/child.py
def negate(self) -> HConfigChild:
"""Negate self.text using driver-specific negation rules.
Negation is resolved in the following priority order:
1. ``negate_with`` rule — replaces ``self.text`` with a custom
negation string defined in the driver (e.g. ``no ip route``).
2. ``negation_default_when`` rule — rewrites the command to its
``default`` form (e.g. ``no shutdown`` → ``default shutdown``).
3. ``negation_sub`` rule — applies a regex substitution to the
negated text (e.g. truncating after a specific token).
4. ``swap_negation`` — toggles the negation prefix/declaration
prefix (e.g. ``shutdown`` ↔ ``no shutdown``).
Returns self so that callers can chain further operations.
"""
if negate_with := self.driver.negate_with(self):
self.text = negate_with
return self
if self.use_default_for_negation(self):
return self._default()
for rule in self.driver.rules.negation_sub:
if self.is_lineage_match(rule.match_rules):
self.text = sub(
rule.search,
rule.replace,
f"{self.driver.negation_prefix}{self.text}",
)
return self
return self.driver.swap_negation(self)
overwrite_with(target, delta, *, negate=True)
Overwrite self's section in delta with a deep copy of target.
When the children of self and target differ, this method mutates
delta in-place: the existing entry for self.text is negated
(if negate=True) or simply deleted (if negate=False), and a
fresh deep copy of target is appended. A "re-create section"
comment is attached to the new entry, and a "dropping section"
comment is added to the negated entry when applicable.
Used by :meth:_config_to_get_to_right when a sectional-overwrite
rule is active for self.text.
Source code in hier_config/child.py
def overwrite_with(
self,
target: HConfigChild,
delta: HConfig | HConfigChild,
*,
negate: bool = True,
) -> None:
"""Overwrite self's section in delta with a deep copy of target.
When the children of self and target differ, this method mutates
``delta`` in-place: the existing entry for ``self.text`` is negated
(if ``negate=True``) or simply deleted (if ``negate=False``), and a
fresh deep copy of ``target`` is appended. A ``"re-create section"``
comment is attached to the new entry, and a ``"dropping section"``
comment is added to the negated entry when applicable.
Used by :meth:`_config_to_get_to_right` when a sectional-overwrite
rule is active for ``self.text``.
"""
if self.children != target.children:
if negate:
if negated := delta.children.get(self.text):
negated.negate()
else:
negated = delta.add_child(
self.text, check_if_present=False
).negate()
negated.comments.add("dropping section")
else:
delta.children.delete(self.text)
if self.children:
new_item = delta.add_deep_copy_of(target)
new_item.comments.add("re-create section")
path()
Yields the text attribute of child objects up to, but excluding, the root.
Source code in hier_config/child.py
def path(self) -> Iterator[str]:
"""Yields the text attribute of child objects up to, but excluding, the root."""
for child in self.lineage():
yield child.text
tags_add(tag)
Add a tag to self._tags on all leaf nodes.
Source code in hier_config/child.py
def tags_add(self, tag: str | Iterable[str]) -> None:
"""Add a tag to self._tags on all leaf nodes."""
if self.is_branch:
for child in self.children:
child.tags_add(tag)
elif isinstance(tag, str):
self._tags.add(tag)
else:
self._tags.update(tag)
tags_remove(tag)
Remove a tag from self._tags on all leaf nodes.
Source code in hier_config/child.py
def tags_remove(self, tag: str | Iterable[str]) -> None:
"""Remove a tag from self._tags on all leaf nodes."""
if self.is_branch:
for child in self.children:
child.tags_remove(tag)
elif isinstance(tag, str):
self._tags.remove(tag)
else:
self._tags.difference_update(tag)
use_sectional_overwrite()
Determines if self.text matches a sectional overwrite rule.
Source code in hier_config/child.py
def use_sectional_overwrite(self) -> bool:
"""Determines if self.text matches a sectional overwrite rule."""
return any(
self.is_lineage_match(rule.match_rules)
for rule in self.driver.rules.sectional_overwrite
)
use_sectional_overwrite_without_negation()
Check self's text to see if negation should be handled by overwriting the section without first negating it.
Source code in hier_config/child.py
def use_sectional_overwrite_without_negation(self) -> bool:
"""Check self's text to see if negation should be handled by
overwriting the section without first negating it.
"""
return any(
self.is_lineage_match(rule.match_rules)
for rule in self.driver.rules.sectional_overwrite_no_negate
)
hier_config.children.HConfigChildren
Ordered collection of HConfigChild objects with fast text-keyed look-up.
Internally maintains both a list (for ordered iteration) and a dict
(for O(1) membership and retrieval by child.text). When duplicate child
text is allowed by the driver, the mapping always points to the first
occurrence while the list preserves all entries in insertion order.
Source code in hier_config/children.py
class HConfigChildren:
"""Ordered collection of `HConfigChild` objects with fast text-keyed look-up.
Internally maintains both a `list` (for ordered iteration) and a `dict`
(for O(1) membership and retrieval by `child.text`). When duplicate child
text is allowed by the driver, the mapping always points to the *first*
occurrence while the list preserves all entries in insertion order.
"""
def __init__(self) -> None:
self._data: list[HConfigChild] = []
self._mapping: dict[str, HConfigChild] = {}
@overload
def __getitem__(self, subscript: int | str) -> HConfigChild: ...
@overload
def __getitem__(self, subscript: slice) -> list[HConfigChild]: ...
def __getitem__(
self, subscript: slice | int | str
) -> HConfigChild | list[HConfigChild]:
if isinstance(subscript, slice):
return self._data[subscript]
if isinstance(subscript, int):
return self._data[subscript]
return self._mapping[subscript]
def __setitem__(self, index: int, child: HConfigChild) -> None:
self._data[index] = child
self.rebuild_mapping()
def __contains__(self, item: str) -> bool:
return item in self._mapping
def __iter__(self) -> Iterator[HConfigChild]:
return iter(self._data)
def __len__(self) -> int:
return len(self._data)
def __eq__(self, other: object) -> bool:
if not isinstance(other, HConfigChildren):
return NotImplemented
self_len = len(self._data)
other_len = len(other._data)
# Superfast succeed method for no children
if self_len == other_len == 0:
return True
# Superfast fail method
if self_len != other_len:
return False
# Fast fail method
if self._mapping.keys() != other._mapping.keys():
return False
# Slower full comparison
return all(
self_child == other_child
for self_child, other_child in zip(
sorted(self._data),
sorted(other._data),
strict=False,
)
)
def __hash__(self) -> int:
return hash(
(*self._data,),
)
def append(
self,
child: HConfigChild,
*,
update_mapping: bool = True,
) -> HConfigChild:
self._data.append(child)
if update_mapping:
self._mapping.setdefault(child.text, child)
return child
def clear(self) -> None:
"""Delete all children."""
self._data.clear()
self._mapping.clear()
def delete(self, child_or_text: HConfigChild | str) -> None:
"""Delete a child from self._data and self._mapping."""
if isinstance(child_or_text, str):
if child_or_text in self._mapping:
self._data[:] = [c for c in self._data if c.text != child_or_text]
self.rebuild_mapping()
else:
old_len = len(self._data)
self._data = [c for c in self._data if c is not child_or_text]
if old_len != len(self._data):
self.rebuild_mapping()
def extend(self, children: Iterable[HConfigChild]) -> None:
"""Add child instances of HConfigChild."""
self._data.extend(children)
for child in children:
self._mapping.setdefault(child.text, child)
def get(self, key: str, default: _D | None = None) -> HConfigChild | _D | None:
return self._mapping.get(key, default)
def index(self, child: HConfigChild) -> int:
return self._data.index(child)
def rebuild_mapping(self) -> None:
"""Rebuild self._mapping."""
self._mapping.clear()
for child in self._data:
self._mapping.setdefault(child.text, child)
clear()
Delete all children.
Source code in hier_config/children.py
def clear(self) -> None:
"""Delete all children."""
self._data.clear()
self._mapping.clear()
delete(child_or_text)
Delete a child from self._data and self._mapping.
Source code in hier_config/children.py
def delete(self, child_or_text: HConfigChild | str) -> None:
"""Delete a child from self._data and self._mapping."""
if isinstance(child_or_text, str):
if child_or_text in self._mapping:
self._data[:] = [c for c in self._data if c.text != child_or_text]
self.rebuild_mapping()
else:
old_len = len(self._data)
self._data = [c for c in self._data if c is not child_or_text]
if old_len != len(self._data):
self.rebuild_mapping()
extend(children)
Add child instances of HConfigChild.
Source code in hier_config/children.py
def extend(self, children: Iterable[HConfigChild]) -> None:
"""Add child instances of HConfigChild."""
self._data.extend(children)
for child in children:
self._mapping.setdefault(child.text, child)
rebuild_mapping()
Rebuild self._mapping.
Source code in hier_config/children.py
def rebuild_mapping(self) -> None:
"""Rebuild self._mapping."""
self._mapping.clear()
for child in self._data:
self._mapping.setdefault(child.text, child)
Workflow
hier_config.WorkflowRemediation
Manages configuration workflows for a network device by comparing running and generated configurations and creating remediations to align the device with the intended configuration state.
| Attributes: |
|---|
| Raises: |
|
|---|
Example
Initialize WorkflowRemediation with the running and generated configurations
and generate remediation and rollback configurations.
from hier_config import WorkflowRemediation, get_hconfig
from hier_config.model import Platform
# Create running and generated configurations as HConfig objects
running_config = get_hconfig(Platform.CISCO_IOS, "running_config_text")
generated_config = get_hconfig(Platform.CISCO_IOS, "generated_config_text")
# Initialize WorkflowRemediation with running and generated configurations
workflow = WorkflowRemediation(running_config, generated_config)
# Generate the remediation configuration to apply the target configuration to the device
remediation_config = workflow.remediation_config
print("Remediation configuration:")
for line in remediation_config.all_children_sorted():
print(line.cisco_style_text())
# Generate the rollback configuration to revert back to the running configuration
rollback_config = workflow.rollback_config
print("Rollback configuration:")
for line in rollback_config.all_children_sorted():
print(line.cisco_style_text())
Source code in hier_config/workflows.py
class WorkflowRemediation:
"""Manages configuration workflows for a network device by comparing
running and generated configurations and creating remediations to align
the device with the intended configuration state.
Attributes:
running_config (HConfig): The current configuration of the network device.
generated_config (HConfig): The target configuration for the network device.
Raises:
ValueError: If `running_config` and `generated_config` have different drivers.
Example:
Initialize `WorkflowRemediation` with the running and generated configurations
and generate remediation and rollback configurations.
```python
from hier_config import WorkflowRemediation, get_hconfig
from hier_config.model import Platform
# Create running and generated configurations as HConfig objects
running_config = get_hconfig(Platform.CISCO_IOS, "running_config_text")
generated_config = get_hconfig(Platform.CISCO_IOS, "generated_config_text")
# Initialize WorkflowRemediation with running and generated configurations
workflow = WorkflowRemediation(running_config, generated_config)
# Generate the remediation configuration to apply the target configuration to the device
remediation_config = workflow.remediation_config
print("Remediation configuration:")
for line in remediation_config.all_children_sorted():
print(line.cisco_style_text())
# Generate the rollback configuration to revert back to the running configuration
rollback_config = workflow.rollback_config
print("Rollback configuration:")
for line in rollback_config.all_children_sorted():
print(line.cisco_style_text())
```
"""
def __init__(
self,
running_config: HConfig,
generated_config: HConfig,
) -> None:
self.running_config = running_config
self.generated_config = generated_config
if running_config.driver.__class__ is not generated_config.driver.__class__:
message = "The running and generated configs must use the same driver."
raise ValueError(message)
self._remediation_config: HConfig | None = None
self._rollback_config: HConfig | None = None
@property
def remediation_config(self) -> HConfig:
"""Builds and returns the remediation configuration to bring the device
in line with the generated configuration.
Returns:
HConfig: The configuration needed to remediate the device.
Notes:
The remediation configuration is cached after the first call.
"""
if self._remediation_config:
return self._remediation_config
remediation_config = self.running_config.config_to_get_to(
self.generated_config
).set_order_weight()
self._remediation_config = remediation_config
return self._remediation_config
@property
def rollback_config(self) -> HConfig:
"""Builds and returns the rollback configuration to revert the device
from the generated configuration back to the running configuration.
Returns:
HConfig: The configuration required to roll back to the original state.
Notes:
The rollback configuration is cached after the first call.
"""
if self._rollback_config:
return self._rollback_config
rollback_config = self.generated_config.config_to_get_to(
self.running_config, HConfig(self.running_config.driver)
).set_order_weight()
self._rollback_config = rollback_config
return rollback_config
def apply_remediation_tag_rules(self, tag_rules: tuple[TagRule, ...]) -> None:
"""Applies tag rules to selectively label parts of the remediation configuration.
Args:
tag_rules (tuple[TagRule, ...]): A set of tag rules specifying sections to tag.
Notes:
This method is useful for managing configuration changes by marking specific
parts of the config for conditional remediation.
"""
for tag_rule in tag_rules:
for child in self.remediation_config.get_children_deep(
tag_rule.match_rules
):
child.tags_add(tag_rule.apply_tags)
def remediation_config_filtered_text(
self,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> str:
"""Returns the remediation configuration as text, filtered by included and excluded tags.
Args:
include_tags (Iterable[str], optional): Tags to include in the output.
exclude_tags (Iterable[str], optional): Tags to exclude from the output.
Returns:
str: The filtered remediation configuration in a text format.
Notes:
- If no tags are provided, the complete sorted remediation configuration is returned.
- Sorting respects configuration hierarchy and specified tags.
"""
children = (
self.remediation_config.all_children_sorted_by_tags(
include_tags, exclude_tags
)
if include_tags or exclude_tags
else self.remediation_config.all_children_sorted()
)
return "\n".join(c.cisco_style_text() for c in children)
remediation_config
property
Builds and returns the remediation configuration to bring the device in line with the generated configuration.
| Returns: |
|
|---|
Notes
The remediation configuration is cached after the first call.
rollback_config
property
Builds and returns the rollback configuration to revert the device from the generated configuration back to the running configuration.
| Returns: |
|
|---|
Notes
The rollback configuration is cached after the first call.
apply_remediation_tag_rules(tag_rules)
Applies tag rules to selectively label parts of the remediation configuration.
| Parameters: |
|
|---|
Notes
This method is useful for managing configuration changes by marking specific parts of the config for conditional remediation.
Source code in hier_config/workflows.py
def apply_remediation_tag_rules(self, tag_rules: tuple[TagRule, ...]) -> None:
"""Applies tag rules to selectively label parts of the remediation configuration.
Args:
tag_rules (tuple[TagRule, ...]): A set of tag rules specifying sections to tag.
Notes:
This method is useful for managing configuration changes by marking specific
parts of the config for conditional remediation.
"""
for tag_rule in tag_rules:
for child in self.remediation_config.get_children_deep(
tag_rule.match_rules
):
child.tags_add(tag_rule.apply_tags)
remediation_config_filtered_text(include_tags=(), exclude_tags=())
Returns the remediation configuration as text, filtered by included and excluded tags.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Notes
- If no tags are provided, the complete sorted remediation configuration is returned.
- Sorting respects configuration hierarchy and specified tags.
Source code in hier_config/workflows.py
def remediation_config_filtered_text(
self,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> str:
"""Returns the remediation configuration as text, filtered by included and excluded tags.
Args:
include_tags (Iterable[str], optional): Tags to include in the output.
exclude_tags (Iterable[str], optional): Tags to exclude from the output.
Returns:
str: The filtered remediation configuration in a text format.
Notes:
- If no tags are provided, the complete sorted remediation configuration is returned.
- Sorting respects configuration hierarchy and specified tags.
"""
children = (
self.remediation_config.all_children_sorted_by_tags(
include_tags, exclude_tags
)
if include_tags or exclude_tags
else self.remediation_config.all_children_sorted()
)
return "\n".join(c.cisco_style_text() for c in children)
Reporting
hier_config.RemediationReporter
A reporting tool for aggregating and analyzing remediation configurations.
This class provides methods to merge multiple device remediations, generate statistics, filter by tags, and export reports in various formats.
Example
from hier_config import RemediationReporter
reporter = RemediationReporter()
reporter.add_remediations([device1_remediation, device2_remediation])
# Get summary statistics
summary = reporter.summary()
# Export reports
reporter.to_json("report.json")
reporter.to_csv("report.csv")
Source code in hier_config/reporting.py
class RemediationReporter: # ruff:ignore[too-many-public-methods]
"""A reporting tool for aggregating and analyzing remediation configurations.
This class provides methods to merge multiple device remediations,
generate statistics, filter by tags, and export reports in various formats.
Example:
```python
from hier_config import RemediationReporter
reporter = RemediationReporter()
reporter.add_remediations([device1_remediation, device2_remediation])
# Get summary statistics
summary = reporter.summary()
# Export reports
reporter.to_json("report.json")
reporter.to_csv("report.csv")
```
"""
def __init__(self) -> None:
"""Initialize a new RemediationReporter."""
self._merged_config: HConfig | None = None
self._device_count: int = 0
self._device_ids: set[int] = set()
@property
def merged_config(self) -> HConfig:
"""The merged configuration.
Raises:
ValueError: If no remediations have been added yet.
"""
if self._merged_config is None:
msg = "No remediations have been added yet"
raise ValueError(msg)
return self._merged_config
@property
def device_count(self) -> int:
"""The number of unique devices that have been added."""
return self._device_count
def add_remediation(self, remediation: HConfig) -> None:
"""Add a single remediation configuration to the reporter.
Args:
remediation: An HConfig object representing a device remediation.
"""
device_id = id(remediation)
if device_id not in self._device_ids:
self._device_ids.add(device_id)
self._device_count += 1
if self._merged_config is None:
# Create a new empty HConfig with the same driver
self._merged_config = HConfig(remediation.driver)
self._merged_config.merge(remediation)
def add_remediations(self, remediations: Iterable[HConfig]) -> None:
"""Add multiple remediation configurations to the reporter.
Args:
remediations: An iterable of HConfig objects.
"""
for remediation in remediations:
self.add_remediation(remediation)
@classmethod
def from_remediations(
cls,
remediations: Iterable[HConfig],
) -> "RemediationReporter":
"""Create a RemediationReporter from an iterable of remediations.
Args:
remediations: An iterable of HConfig objects.
Returns:
A new RemediationReporter instance with all remediations merged.
Example:
```python
reporter = RemediationReporter.from_remediations(
[
device1_remediation,
device2_remediation,
]
)
```
"""
reporter = cls()
reporter.add_remediations(remediations)
return reporter
@classmethod
def from_merged_config(cls, merged_config: HConfig) -> "RemediationReporter":
"""Create a RemediationReporter from an already merged configuration.
Args:
merged_config: An HConfig object with merged remediations.
Returns:
A new RemediationReporter instance.
Example:
```python
merged = get_hconfig(Platform.CISCO_IOS)
merged.merge([device1, device2])
reporter = RemediationReporter.from_merged_config(merged)
```
"""
reporter = cls()
reporter._merged_config = merged_config
# Count unique device IDs from instances
device_ids: set[int] = set()
for child in merged_config.all_children_sorted():
device_ids.update(instance.id for instance in child.instances)
reporter._device_ids = device_ids
reporter._device_count = len(device_ids)
return reporter
def apply_tag_rules(self, tag_rules: Sequence[TagRule]) -> None:
"""Apply tag rules to the merged configuration.
Args:
tag_rules: A sequence of TagRule objects to apply.
Example:
```python
from hier_config import MatchRule, TagRule
tag_rules = [
TagRule(
match_rules=(MatchRule(startswith="ntp"),),
apply_tags=frozenset({"ntp", "safe"}),
)
]
reporter.apply_tag_rules(tag_rules)
```
"""
for tag_rule in tag_rules:
for child in self.merged_config.get_children_deep(tag_rule.match_rules):
child.tags_add(tag_rule.apply_tags)
def get_all_changes(
self,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[HConfigChild, ...]:
"""Get all configuration changes, optionally filtered by tags.
Args:
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of HConfigChild objects representing changes.
"""
if include_tags or exclude_tags:
return tuple(
self.merged_config.all_children_sorted_by_tags(
include_tags,
exclude_tags,
)
)
return tuple(self.merged_config.all_children_sorted())
def get_change_detail(
self,
line: str,
*,
tag: str | None = None,
) -> ChangeDetail | None:
"""Get detailed information about a specific configuration line.
Args:
line: The configuration line to search for.
tag: Optional tag to filter instances.
Returns:
A ChangeDetail object or None if the line is not found.
Example:
```python
detail = reporter.get_change_detail("line vty 0 4")
print(f"Affects {detail.device_count} devices")
```
"""
child = self.merged_config.get_child(equals=line)
if child is None:
return None
return self._build_change_detail(child, tag=tag)
def _build_change_detail(
self,
child: HConfigChild,
*,
tag: str | None = None,
) -> ChangeDetail:
"""Build a ChangeDetail object from an HConfigChild.
Args:
child: The HConfigChild to build details from.
tag: Optional tag to filter instances.
Returns:
A ChangeDetail object with aggregated information.
"""
# Filter instances by tag if specified
relevant_instances = tuple(
instance
for instance in child.instances
if tag is None or tag in instance.tags
)
# Aggregate data
device_ids = frozenset(instance.id for instance in relevant_instances)
all_tags = frozenset(
tag_item for instance in relevant_instances for tag_item in instance.tags
)
all_comments = frozenset(
comment for instance in relevant_instances for comment in instance.comments
)
# Build path
path_parts: list[str] = []
current: HConfigChild | HConfig | None = child
while current is not None and hasattr(current, "text"):
if isinstance(current, HConfigChild):
path_parts.insert(0, current.text)
current = getattr(current, "parent", None)
# Get children details
children_details = tuple(
self._build_change_detail(grandchild, tag=tag)
for grandchild in child.children
)
return ChangeDetail(
line=child.text,
full_path=tuple(path_parts),
device_count=len(device_ids),
device_ids=device_ids,
tags=all_tags,
comments=all_comments,
instances=relevant_instances,
children=children_details,
)
def get_device_count(self, line: str, *, tag: str | None = None) -> int:
"""The number of devices that need a specific configuration line.
Args:
line: The configuration line to search for.
tag: Optional tag to filter instances.
Returns:
The number of devices requiring this change.
Example:
```python
count = reporter.get_device_count("line vty 0 4")
print(f"{count} devices need this change")
```
"""
detail = self.get_change_detail(line, tag=tag)
return detail.device_count if detail else 0
def get_changes_by_threshold(
self,
*,
min_devices: int = 0,
max_devices: int | None = None,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[HConfigChild, ...]:
"""Get changes affecting a certain number of devices.
Args:
min_devices: Minimum number of devices (inclusive).
max_devices: Maximum number of devices (inclusive), or None for unlimited.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of HConfigChild objects matching the criteria.
Example:
```python
# Get high-impact changes affecting 50+ devices
high_impact = reporter.get_changes_by_threshold(min_devices=50)
# Get isolated changes affecting 1-5 devices
isolated = reporter.get_changes_by_threshold(
min_devices=1,
max_devices=5,
)
```
"""
all_changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
filtered_changes: list[HConfigChild] = []
for child in all_changes:
instance_count = len(child.instances)
if instance_count >= min_devices and (
max_devices is None or instance_count <= max_devices
):
filtered_changes.append(child)
result: tuple[HConfigChild, ...] = tuple(filtered_changes)
return result
def get_top_changes(
self,
n: int = 10,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[tuple[HConfigChild, int], ...]:
"""The top N most common changes across devices.
Args:
n: Number of top changes to return.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of (HConfigChild, count) pairs, sorted by count descending.
Example:
```python
top_10 = reporter.get_top_changes(10)
for child, count in top_10:
print(f"{child.text}: {count} devices")
```
"""
all_changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
# Create list of (child, instance_count) pairs
changes_with_counts = [(child, len(child.instances)) for child in all_changes]
# Sort by count descending and take top N
changes_with_counts.sort(key=operator.itemgetter(1), reverse=True)
return tuple(changes_with_counts[:n])
def get_changes_matching(
self,
pattern: str,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[HConfigChild, ...]:
r"""Get changes matching a regex pattern.
Args:
pattern: A regex pattern to match against configuration lines.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of HConfigChild objects matching the pattern.
Example:
```python
# Get all VLAN interface changes
vlan_changes = reporter.get_changes_matching(r"interface Vlan\\d+")
```
"""
all_changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
regex = re.compile(pattern)
return tuple(child for child in all_changes if regex.search(child.text))
def summary(self) -> ReportSummary:
"""Generate a summary of all remediations.
Returns:
A ReportSummary object with aggregate statistics.
Example:
```python
summary = reporter.summary()
print(f"Total devices: {summary.total_devices}")
print(f"Unique changes: {summary.total_unique_changes}")
```
"""
all_changes = self.get_all_changes()
# Get most common changes
changes_with_counts = [
(child.text, len(child.instances)) for child in all_changes
]
changes_with_counts.sort(key=operator.itemgetter(1), reverse=True)
most_common = tuple(changes_with_counts[:10])
# Count changes by tag
tag_counter: Counter[str] = Counter()
for child in all_changes:
for tag in child.tags:
tag_counter[tag] += 1
return ReportSummary(
total_devices=self.device_count,
total_unique_changes=len(all_changes),
most_common_changes=most_common,
changes_by_tag=dict(tag_counter),
)
def summary_by_tags(
self,
tags: Iterable[str] | None = None,
) -> dict[str, dict[str, Any]]:
"""Generate a summary breakdown by tags.
Args:
tags: Specific tags to include, or None for all tags.
Returns:
A dictionary mapping tag names to their statistics.
Example:
```python
tag_summary = reporter.summary_by_tags(["security", "ntp"])
for tag, stats in tag_summary.items():
print(f"{tag}: {stats['device_count']} devices")
```
"""
all_changes = self.get_all_changes()
# Collect all tags if not specified
all_tags = (
{tag for child in all_changes for tag in child.tags}
if tags is None
else set(tags)
)
result: dict[str, dict[str, Any]] = {}
for tag in all_tags:
tagged_changes = self.get_all_changes(include_tags=[tag])
# Count unique devices for this tag
device_ids = {
instance.id
for child in tagged_changes
for instance in child.instances
if tag in instance.tags
}
result[tag] = {
"device_count": len(device_ids),
"change_count": len(tagged_changes),
"changes": [child.text for child in tagged_changes],
}
final_result: dict[str, dict[str, Any]] = result
return final_result
def summary_text(self, *, top_n: int = 10) -> str:
"""Generate a human-readable text summary.
Args:
top_n: Number of top changes to include in the summary.
Returns:
A formatted string with summary statistics.
Example:
```python
print(reporter.summary_text())
```
"""
summary = self.summary()
lines = [
"Remediation Summary",
"=" * 50,
f"Total devices: {summary.total_devices}",
f"Unique changes: {summary.total_unique_changes}",
"",
f"Top {min(top_n, len(summary.most_common_changes))} Most Common Changes:",
"-" * 50,
]
for i, (line, count) in enumerate(summary.most_common_changes[:top_n], 1):
percentage = (
(count / summary.total_devices * 100) if summary.total_devices else 0
)
lines.extend((f"{i}. {line}", f" {count} devices ({percentage:.1f}%)"))
if summary.changes_by_tag:
lines.extend(["", "Changes by Tag:", "-" * 50])
for tag, count in sorted(
summary.changes_by_tag.items(),
key=operator.itemgetter(1),
reverse=True,
):
lines.append(f" {tag}: {count} changes")
return "\n".join(lines)
def group_by_parent(self) -> dict[str, list[HConfigChild]]:
"""Group all changes by their parent configuration line.
Returns:
A dictionary mapping parent lines to their children.
Example:
```python
grouped = reporter.group_by_parent()
for parent, children in grouped.items():
print(f"{parent}: {len(children)} changes")
```
"""
groups: dict[str, list[HConfigChild]] = defaultdict(list)
for child in self.get_all_changes():
parent_text = (
child.parent.text if isinstance(child.parent, HConfigChild) else "root"
)
groups[parent_text].append(child)
return dict(groups)
def get_impact_distribution(
self,
bins: Sequence[int] = (1, 10, 25, 50, 100),
) -> dict[str, int]:
"""The distribution of changes by device impact.
Args:
bins: Boundaries for impact ranges.
Returns:
A dictionary with range labels as keys and counts as values.
Example:
```python
dist = reporter.get_impact_distribution()
# {'1-10': 15, '10-25': 8, '25-50': 5, '50-100': 3, '100+': 2}
```
"""
all_changes = self.get_all_changes()
distribution: dict[str, int] = defaultdict(int)
for child in all_changes:
instance_count = len(child.instances)
# Find the appropriate bin
for i, threshold in enumerate(bins):
if instance_count < threshold:
label = f"1-{threshold}" if i == 0 else f"{bins[i - 1]}-{threshold}"
distribution[label] += 1
break
else:
# Higher than all bins
distribution[f"{bins[-1]}+"] += 1
return dict(distribution)
def get_tag_distribution(self) -> dict[str, int]:
"""The distribution of tags across all changes.
Returns:
A dictionary mapping tag names to their occurrence count.
Example:
```python
tags = reporter.get_tag_distribution()
# {'security': 45, 'ntp': 32, 'snmp': 28}
```
"""
tag_counter: Counter[str] = Counter()
for child in self.get_all_changes():
for tag in child.tags:
tag_counter[tag] += 1
return dict(tag_counter)
def to_text(
self,
file_path: str | Path,
*,
style: TextStyle = "merged",
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> None:
"""Export remediation configuration to a text file.
Args:
file_path: Path to the output file.
style: Text style ('merged', 'with_comments', or 'without_comments').
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Example:
```python
reporter.to_text("remediation.txt", style="merged")
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
lines = [child.cisco_style_text(style=style) for child in changes]
output_path = Path(file_path)
output_path.write_text("\n".join(lines), encoding="utf-8")
def to_json(
self,
file_path: str | Path,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
indent: int = 2,
) -> None:
"""Export remediation data to a JSON file.
Args:
file_path: Path to the output file.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
indent: JSON indentation level.
Example:
```python
reporter.to_json("remediation.json")
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
data = {
"summary": {
"total_devices": self.device_count,
"total_unique_changes": len(changes),
},
"changes": [
{
"line": child.text,
"device_count": len(child.instances),
"device_ids": sorted(instance.id for instance in child.instances),
"tags": sorted(child.tags),
"comments": sorted(child.comments),
"instances": [
{
"id": instance.id,
"tags": sorted(instance.tags),
"comments": sorted(instance.comments),
}
for instance in child.instances
],
}
for child in changes
],
}
output_path = Path(file_path)
output_path.write_text(
json.dumps(data, indent=indent),
encoding="utf-8",
)
def to_csv(
self,
file_path: str | Path,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> None:
"""Export remediation data to a CSV file.
Args:
file_path: Path to the output file.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Example:
```python
reporter.to_csv("remediation.csv")
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
output_path = Path(file_path)
with output_path.open("w", encoding="utf-8", newline="") as csvfile:
writer = csv.writer(csvfile)
writer.writerow(
[
"line",
"device_count",
"percentage",
"tags",
"comments",
"device_ids",
]
)
for child in changes:
device_count = len(child.instances)
percentage = (
(device_count / self.device_count * 100) if self.device_count else 0
)
tags = ",".join(sorted(child.tags))
comments = " | ".join(sorted(child.comments))
device_ids = ",".join(str(instance.id) for instance in child.instances)
writer.writerow(
[
child.text,
device_count,
f"{percentage:.1f}",
tags,
comments,
device_ids,
]
)
def to_markdown(
self,
file_path: str | Path,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
top_n: int = 20,
) -> None:
"""Export remediation report to a Markdown file.
Args:
file_path: Path to the output file.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
top_n: Number of top changes to include in detail.
Example:
```python
reporter.to_markdown("remediation.md", top_n=15)
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
lines = [
"# Remediation Report",
"",
"## Summary",
"",
f"- **Total Devices**: {self.device_count}",
f"- **Unique Changes**: {len(changes)}",
"",
f"## Top {min(top_n, len(changes))} Changes by Impact",
"",
"| # | Configuration Line | Device Count | Percentage |",
"|---|-------------------|--------------|------------|",
]
# Sort by instance count
sorted_changes = sorted(
changes,
key=lambda c: len(c.instances),
reverse=True,
)
for i, child in enumerate(sorted_changes[:top_n], 1):
device_count = len(child.instances)
percentage = (
(device_count / self.device_count * 100) if self.device_count else 0
)
lines.append(
f"| {i} | `{child.text}` | {device_count} | {percentage:.1f}% |"
)
# Add tag summary if tags exist
tag_dist = self.get_tag_distribution()
if tag_dist:
lines.extend(
[
"",
"## Changes by Tag",
"",
"| Tag | Count |",
"|-----|-------|",
]
)
for tag, count in sorted(
tag_dist.items(),
key=operator.itemgetter(1),
reverse=True,
):
lines.append(f"| {tag} | {count} |")
output_path = Path(file_path)
output_path.write_text("\n".join(lines), encoding="utf-8")
def export_all(
self,
output_dir: str | Path,
*,
formats: Iterable[str] = ("json", "csv", "markdown", "text"),
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> None:
"""Export reports in multiple formats to a directory.
Args:
output_dir: Directory to write reports to.
formats: Formats to export ('json', 'csv', 'markdown', 'text').
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Example:
```python
reporter.export_all(
"reports/",
formats=["json", "csv", "markdown"],
)
```
"""
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
# Map format names to file extensions
ext_map = {"json": "json", "csv": "csv", "markdown": "md", "text": "txt"}
for fmt in formats:
file_ext = ext_map.get(fmt, fmt)
file_path_str = str(output_path / f"remediation_report.{file_ext}")
if fmt == "json":
self.to_json(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
elif fmt == "csv":
self.to_csv(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
elif fmt == "markdown":
self.to_markdown(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
elif fmt == "text":
self.to_text(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
device_count
property
The number of unique devices that have been added.
merged_config
property
The merged configuration.
| Raises: |
|
|---|
__init__()
Initialize a new RemediationReporter.
Source code in hier_config/reporting.py
def __init__(self) -> None:
"""Initialize a new RemediationReporter."""
self._merged_config: HConfig | None = None
self._device_count: int = 0
self._device_ids: set[int] = set()
add_remediation(remediation)
Add a single remediation configuration to the reporter.
| Parameters: |
|
|---|
Source code in hier_config/reporting.py
def add_remediation(self, remediation: HConfig) -> None:
"""Add a single remediation configuration to the reporter.
Args:
remediation: An HConfig object representing a device remediation.
"""
device_id = id(remediation)
if device_id not in self._device_ids:
self._device_ids.add(device_id)
self._device_count += 1
if self._merged_config is None:
# Create a new empty HConfig with the same driver
self._merged_config = HConfig(remediation.driver)
self._merged_config.merge(remediation)
add_remediations(remediations)
Add multiple remediation configurations to the reporter.
| Parameters: |
|
|---|
Source code in hier_config/reporting.py
def add_remediations(self, remediations: Iterable[HConfig]) -> None:
"""Add multiple remediation configurations to the reporter.
Args:
remediations: An iterable of HConfig objects.
"""
for remediation in remediations:
self.add_remediation(remediation)
apply_tag_rules(tag_rules)
Apply tag rules to the merged configuration.
| Parameters: |
|
|---|
Example
from hier_config import MatchRule, TagRule
tag_rules = [
TagRule(
match_rules=(MatchRule(startswith="ntp"),),
apply_tags=frozenset({"ntp", "safe"}),
)
]
reporter.apply_tag_rules(tag_rules)
Source code in hier_config/reporting.py
def apply_tag_rules(self, tag_rules: Sequence[TagRule]) -> None:
"""Apply tag rules to the merged configuration.
Args:
tag_rules: A sequence of TagRule objects to apply.
Example:
```python
from hier_config import MatchRule, TagRule
tag_rules = [
TagRule(
match_rules=(MatchRule(startswith="ntp"),),
apply_tags=frozenset({"ntp", "safe"}),
)
]
reporter.apply_tag_rules(tag_rules)
```
"""
for tag_rule in tag_rules:
for child in self.merged_config.get_children_deep(tag_rule.match_rules):
child.tags_add(tag_rule.apply_tags)
export_all(output_dir, *, formats=('json', 'csv', 'markdown', 'text'), include_tags=(), exclude_tags=())
Export reports in multiple formats to a directory.
| Parameters: |
|
|---|
Example
reporter.export_all(
"reports/",
formats=["json", "csv", "markdown"],
)
Source code in hier_config/reporting.py
def export_all(
self,
output_dir: str | Path,
*,
formats: Iterable[str] = ("json", "csv", "markdown", "text"),
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> None:
"""Export reports in multiple formats to a directory.
Args:
output_dir: Directory to write reports to.
formats: Formats to export ('json', 'csv', 'markdown', 'text').
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Example:
```python
reporter.export_all(
"reports/",
formats=["json", "csv", "markdown"],
)
```
"""
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
# Map format names to file extensions
ext_map = {"json": "json", "csv": "csv", "markdown": "md", "text": "txt"}
for fmt in formats:
file_ext = ext_map.get(fmt, fmt)
file_path_str = str(output_path / f"remediation_report.{file_ext}")
if fmt == "json":
self.to_json(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
elif fmt == "csv":
self.to_csv(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
elif fmt == "markdown":
self.to_markdown(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
elif fmt == "text":
self.to_text(
file_path_str,
include_tags=include_tags,
exclude_tags=exclude_tags,
)
from_merged_config(merged_config)
classmethod
Create a RemediationReporter from an already merged configuration.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
merged = get_hconfig(Platform.CISCO_IOS)
merged.merge([device1, device2])
reporter = RemediationReporter.from_merged_config(merged)
Source code in hier_config/reporting.py
@classmethod
def from_merged_config(cls, merged_config: HConfig) -> "RemediationReporter":
"""Create a RemediationReporter from an already merged configuration.
Args:
merged_config: An HConfig object with merged remediations.
Returns:
A new RemediationReporter instance.
Example:
```python
merged = get_hconfig(Platform.CISCO_IOS)
merged.merge([device1, device2])
reporter = RemediationReporter.from_merged_config(merged)
```
"""
reporter = cls()
reporter._merged_config = merged_config
# Count unique device IDs from instances
device_ids: set[int] = set()
for child in merged_config.all_children_sorted():
device_ids.update(instance.id for instance in child.instances)
reporter._device_ids = device_ids
reporter._device_count = len(device_ids)
return reporter
from_remediations(remediations)
classmethod
Create a RemediationReporter from an iterable of remediations.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
reporter = RemediationReporter.from_remediations(
[
device1_remediation,
device2_remediation,
]
)
Source code in hier_config/reporting.py
@classmethod
def from_remediations(
cls,
remediations: Iterable[HConfig],
) -> "RemediationReporter":
"""Create a RemediationReporter from an iterable of remediations.
Args:
remediations: An iterable of HConfig objects.
Returns:
A new RemediationReporter instance with all remediations merged.
Example:
```python
reporter = RemediationReporter.from_remediations(
[
device1_remediation,
device2_remediation,
]
)
```
"""
reporter = cls()
reporter.add_remediations(remediations)
return reporter
get_all_changes(*, include_tags=(), exclude_tags=())
Get all configuration changes, optionally filtered by tags.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Source code in hier_config/reporting.py
def get_all_changes(
self,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[HConfigChild, ...]:
"""Get all configuration changes, optionally filtered by tags.
Args:
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of HConfigChild objects representing changes.
"""
if include_tags or exclude_tags:
return tuple(
self.merged_config.all_children_sorted_by_tags(
include_tags,
exclude_tags,
)
)
return tuple(self.merged_config.all_children_sorted())
get_change_detail(line, *, tag=None)
Get detailed information about a specific configuration line.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
detail = reporter.get_change_detail("line vty 0 4")
print(f"Affects {detail.device_count} devices")
Source code in hier_config/reporting.py
def get_change_detail(
self,
line: str,
*,
tag: str | None = None,
) -> ChangeDetail | None:
"""Get detailed information about a specific configuration line.
Args:
line: The configuration line to search for.
tag: Optional tag to filter instances.
Returns:
A ChangeDetail object or None if the line is not found.
Example:
```python
detail = reporter.get_change_detail("line vty 0 4")
print(f"Affects {detail.device_count} devices")
```
"""
child = self.merged_config.get_child(equals=line)
if child is None:
return None
return self._build_change_detail(child, tag=tag)
get_changes_by_threshold(*, min_devices=0, max_devices=None, include_tags=(), exclude_tags=())
Get changes affecting a certain number of devices.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
# Get high-impact changes affecting 50+ devices
high_impact = reporter.get_changes_by_threshold(min_devices=50)
# Get isolated changes affecting 1-5 devices
isolated = reporter.get_changes_by_threshold(
min_devices=1,
max_devices=5,
)
Source code in hier_config/reporting.py
def get_changes_by_threshold(
self,
*,
min_devices: int = 0,
max_devices: int | None = None,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[HConfigChild, ...]:
"""Get changes affecting a certain number of devices.
Args:
min_devices: Minimum number of devices (inclusive).
max_devices: Maximum number of devices (inclusive), or None for unlimited.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of HConfigChild objects matching the criteria.
Example:
```python
# Get high-impact changes affecting 50+ devices
high_impact = reporter.get_changes_by_threshold(min_devices=50)
# Get isolated changes affecting 1-5 devices
isolated = reporter.get_changes_by_threshold(
min_devices=1,
max_devices=5,
)
```
"""
all_changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
filtered_changes: list[HConfigChild] = []
for child in all_changes:
instance_count = len(child.instances)
if instance_count >= min_devices and (
max_devices is None or instance_count <= max_devices
):
filtered_changes.append(child)
result: tuple[HConfigChild, ...] = tuple(filtered_changes)
return result
get_changes_matching(pattern, *, include_tags=(), exclude_tags=())
Get changes matching a regex pattern.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
# Get all VLAN interface changes
vlan_changes = reporter.get_changes_matching(r"interface Vlan\\d+")
Source code in hier_config/reporting.py
def get_changes_matching(
self,
pattern: str,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[HConfigChild, ...]:
r"""Get changes matching a regex pattern.
Args:
pattern: A regex pattern to match against configuration lines.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of HConfigChild objects matching the pattern.
Example:
```python
# Get all VLAN interface changes
vlan_changes = reporter.get_changes_matching(r"interface Vlan\\d+")
```
"""
all_changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
regex = re.compile(pattern)
return tuple(child for child in all_changes if regex.search(child.text))
get_device_count(line, *, tag=None)
The number of devices that need a specific configuration line.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
count = reporter.get_device_count("line vty 0 4")
print(f"{count} devices need this change")
Source code in hier_config/reporting.py
def get_device_count(self, line: str, *, tag: str | None = None) -> int:
"""The number of devices that need a specific configuration line.
Args:
line: The configuration line to search for.
tag: Optional tag to filter instances.
Returns:
The number of devices requiring this change.
Example:
```python
count = reporter.get_device_count("line vty 0 4")
print(f"{count} devices need this change")
```
"""
detail = self.get_change_detail(line, tag=tag)
return detail.device_count if detail else 0
get_impact_distribution(bins=(1, 10, 25, 50, 100))
The distribution of changes by device impact.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
dist = reporter.get_impact_distribution()
# {'1-10': 15, '10-25': 8, '25-50': 5, '50-100': 3, '100+': 2}
Source code in hier_config/reporting.py
def get_impact_distribution(
self,
bins: Sequence[int] = (1, 10, 25, 50, 100),
) -> dict[str, int]:
"""The distribution of changes by device impact.
Args:
bins: Boundaries for impact ranges.
Returns:
A dictionary with range labels as keys and counts as values.
Example:
```python
dist = reporter.get_impact_distribution()
# {'1-10': 15, '10-25': 8, '25-50': 5, '50-100': 3, '100+': 2}
```
"""
all_changes = self.get_all_changes()
distribution: dict[str, int] = defaultdict(int)
for child in all_changes:
instance_count = len(child.instances)
# Find the appropriate bin
for i, threshold in enumerate(bins):
if instance_count < threshold:
label = f"1-{threshold}" if i == 0 else f"{bins[i - 1]}-{threshold}"
distribution[label] += 1
break
else:
# Higher than all bins
distribution[f"{bins[-1]}+"] += 1
return dict(distribution)
get_tag_distribution()
The distribution of tags across all changes.
| Returns: |
|
|---|
Example
tags = reporter.get_tag_distribution()
# {'security': 45, 'ntp': 32, 'snmp': 28}
Source code in hier_config/reporting.py
def get_tag_distribution(self) -> dict[str, int]:
"""The distribution of tags across all changes.
Returns:
A dictionary mapping tag names to their occurrence count.
Example:
```python
tags = reporter.get_tag_distribution()
# {'security': 45, 'ntp': 32, 'snmp': 28}
```
"""
tag_counter: Counter[str] = Counter()
for child in self.get_all_changes():
for tag in child.tags:
tag_counter[tag] += 1
return dict(tag_counter)
get_top_changes(n=10, *, include_tags=(), exclude_tags=())
The top N most common changes across devices.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
top_10 = reporter.get_top_changes(10)
for child, count in top_10:
print(f"{child.text}: {count} devices")
Source code in hier_config/reporting.py
def get_top_changes(
self,
n: int = 10,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> tuple[tuple[HConfigChild, int], ...]:
"""The top N most common changes across devices.
Args:
n: Number of top changes to return.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Returns:
A tuple of (HConfigChild, count) pairs, sorted by count descending.
Example:
```python
top_10 = reporter.get_top_changes(10)
for child, count in top_10:
print(f"{child.text}: {count} devices")
```
"""
all_changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
# Create list of (child, instance_count) pairs
changes_with_counts = [(child, len(child.instances)) for child in all_changes]
# Sort by count descending and take top N
changes_with_counts.sort(key=operator.itemgetter(1), reverse=True)
return tuple(changes_with_counts[:n])
group_by_parent()
Group all changes by their parent configuration line.
| Returns: |
|
|---|
Example
grouped = reporter.group_by_parent()
for parent, children in grouped.items():
print(f"{parent}: {len(children)} changes")
Source code in hier_config/reporting.py
def group_by_parent(self) -> dict[str, list[HConfigChild]]:
"""Group all changes by their parent configuration line.
Returns:
A dictionary mapping parent lines to their children.
Example:
```python
grouped = reporter.group_by_parent()
for parent, children in grouped.items():
print(f"{parent}: {len(children)} changes")
```
"""
groups: dict[str, list[HConfigChild]] = defaultdict(list)
for child in self.get_all_changes():
parent_text = (
child.parent.text if isinstance(child.parent, HConfigChild) else "root"
)
groups[parent_text].append(child)
return dict(groups)
summary()
Generate a summary of all remediations.
| Returns: |
|
|---|
Example
summary = reporter.summary()
print(f"Total devices: {summary.total_devices}")
print(f"Unique changes: {summary.total_unique_changes}")
Source code in hier_config/reporting.py
def summary(self) -> ReportSummary:
"""Generate a summary of all remediations.
Returns:
A ReportSummary object with aggregate statistics.
Example:
```python
summary = reporter.summary()
print(f"Total devices: {summary.total_devices}")
print(f"Unique changes: {summary.total_unique_changes}")
```
"""
all_changes = self.get_all_changes()
# Get most common changes
changes_with_counts = [
(child.text, len(child.instances)) for child in all_changes
]
changes_with_counts.sort(key=operator.itemgetter(1), reverse=True)
most_common = tuple(changes_with_counts[:10])
# Count changes by tag
tag_counter: Counter[str] = Counter()
for child in all_changes:
for tag in child.tags:
tag_counter[tag] += 1
return ReportSummary(
total_devices=self.device_count,
total_unique_changes=len(all_changes),
most_common_changes=most_common,
changes_by_tag=dict(tag_counter),
)
summary_by_tags(tags=None)
Generate a summary breakdown by tags.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
tag_summary = reporter.summary_by_tags(["security", "ntp"])
for tag, stats in tag_summary.items():
print(f"{tag}: {stats['device_count']} devices")
Source code in hier_config/reporting.py
def summary_by_tags(
self,
tags: Iterable[str] | None = None,
) -> dict[str, dict[str, Any]]:
"""Generate a summary breakdown by tags.
Args:
tags: Specific tags to include, or None for all tags.
Returns:
A dictionary mapping tag names to their statistics.
Example:
```python
tag_summary = reporter.summary_by_tags(["security", "ntp"])
for tag, stats in tag_summary.items():
print(f"{tag}: {stats['device_count']} devices")
```
"""
all_changes = self.get_all_changes()
# Collect all tags if not specified
all_tags = (
{tag for child in all_changes for tag in child.tags}
if tags is None
else set(tags)
)
result: dict[str, dict[str, Any]] = {}
for tag in all_tags:
tagged_changes = self.get_all_changes(include_tags=[tag])
# Count unique devices for this tag
device_ids = {
instance.id
for child in tagged_changes
for instance in child.instances
if tag in instance.tags
}
result[tag] = {
"device_count": len(device_ids),
"change_count": len(tagged_changes),
"changes": [child.text for child in tagged_changes],
}
final_result: dict[str, dict[str, Any]] = result
return final_result
summary_text(*, top_n=10)
Generate a human-readable text summary.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
print(reporter.summary_text())
Source code in hier_config/reporting.py
def summary_text(self, *, top_n: int = 10) -> str:
"""Generate a human-readable text summary.
Args:
top_n: Number of top changes to include in the summary.
Returns:
A formatted string with summary statistics.
Example:
```python
print(reporter.summary_text())
```
"""
summary = self.summary()
lines = [
"Remediation Summary",
"=" * 50,
f"Total devices: {summary.total_devices}",
f"Unique changes: {summary.total_unique_changes}",
"",
f"Top {min(top_n, len(summary.most_common_changes))} Most Common Changes:",
"-" * 50,
]
for i, (line, count) in enumerate(summary.most_common_changes[:top_n], 1):
percentage = (
(count / summary.total_devices * 100) if summary.total_devices else 0
)
lines.extend((f"{i}. {line}", f" {count} devices ({percentage:.1f}%)"))
if summary.changes_by_tag:
lines.extend(["", "Changes by Tag:", "-" * 50])
for tag, count in sorted(
summary.changes_by_tag.items(),
key=operator.itemgetter(1),
reverse=True,
):
lines.append(f" {tag}: {count} changes")
return "\n".join(lines)
to_csv(file_path, *, include_tags=(), exclude_tags=())
Export remediation data to a CSV file.
| Parameters: |
|
|---|
Example
reporter.to_csv("remediation.csv")
Source code in hier_config/reporting.py
def to_csv(
self,
file_path: str | Path,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> None:
"""Export remediation data to a CSV file.
Args:
file_path: Path to the output file.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Example:
```python
reporter.to_csv("remediation.csv")
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
output_path = Path(file_path)
with output_path.open("w", encoding="utf-8", newline="") as csvfile:
writer = csv.writer(csvfile)
writer.writerow(
[
"line",
"device_count",
"percentage",
"tags",
"comments",
"device_ids",
]
)
for child in changes:
device_count = len(child.instances)
percentage = (
(device_count / self.device_count * 100) if self.device_count else 0
)
tags = ",".join(sorted(child.tags))
comments = " | ".join(sorted(child.comments))
device_ids = ",".join(str(instance.id) for instance in child.instances)
writer.writerow(
[
child.text,
device_count,
f"{percentage:.1f}",
tags,
comments,
device_ids,
]
)
to_json(file_path, *, include_tags=(), exclude_tags=(), indent=2)
Export remediation data to a JSON file.
| Parameters: |
|
|---|
Example
reporter.to_json("remediation.json")
Source code in hier_config/reporting.py
def to_json(
self,
file_path: str | Path,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
indent: int = 2,
) -> None:
"""Export remediation data to a JSON file.
Args:
file_path: Path to the output file.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
indent: JSON indentation level.
Example:
```python
reporter.to_json("remediation.json")
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
data = {
"summary": {
"total_devices": self.device_count,
"total_unique_changes": len(changes),
},
"changes": [
{
"line": child.text,
"device_count": len(child.instances),
"device_ids": sorted(instance.id for instance in child.instances),
"tags": sorted(child.tags),
"comments": sorted(child.comments),
"instances": [
{
"id": instance.id,
"tags": sorted(instance.tags),
"comments": sorted(instance.comments),
}
for instance in child.instances
],
}
for child in changes
],
}
output_path = Path(file_path)
output_path.write_text(
json.dumps(data, indent=indent),
encoding="utf-8",
)
to_markdown(file_path, *, include_tags=(), exclude_tags=(), top_n=20)
Export remediation report to a Markdown file.
| Parameters: |
|
|---|
Example
reporter.to_markdown("remediation.md", top_n=15)
Source code in hier_config/reporting.py
def to_markdown(
self,
file_path: str | Path,
*,
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
top_n: int = 20,
) -> None:
"""Export remediation report to a Markdown file.
Args:
file_path: Path to the output file.
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
top_n: Number of top changes to include in detail.
Example:
```python
reporter.to_markdown("remediation.md", top_n=15)
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
lines = [
"# Remediation Report",
"",
"## Summary",
"",
f"- **Total Devices**: {self.device_count}",
f"- **Unique Changes**: {len(changes)}",
"",
f"## Top {min(top_n, len(changes))} Changes by Impact",
"",
"| # | Configuration Line | Device Count | Percentage |",
"|---|-------------------|--------------|------------|",
]
# Sort by instance count
sorted_changes = sorted(
changes,
key=lambda c: len(c.instances),
reverse=True,
)
for i, child in enumerate(sorted_changes[:top_n], 1):
device_count = len(child.instances)
percentage = (
(device_count / self.device_count * 100) if self.device_count else 0
)
lines.append(
f"| {i} | `{child.text}` | {device_count} | {percentage:.1f}% |"
)
# Add tag summary if tags exist
tag_dist = self.get_tag_distribution()
if tag_dist:
lines.extend(
[
"",
"## Changes by Tag",
"",
"| Tag | Count |",
"|-----|-------|",
]
)
for tag, count in sorted(
tag_dist.items(),
key=operator.itemgetter(1),
reverse=True,
):
lines.append(f"| {tag} | {count} |")
output_path = Path(file_path)
output_path.write_text("\n".join(lines), encoding="utf-8")
to_text(file_path, *, style='merged', include_tags=(), exclude_tags=())
Export remediation configuration to a text file.
| Parameters: |
|
|---|
Example
reporter.to_text("remediation.txt", style="merged")
Source code in hier_config/reporting.py
def to_text(
self,
file_path: str | Path,
*,
style: TextStyle = "merged",
include_tags: Iterable[str] = (),
exclude_tags: Iterable[str] = (),
) -> None:
"""Export remediation configuration to a text file.
Args:
file_path: Path to the output file.
style: Text style ('merged', 'with_comments', or 'without_comments').
include_tags: Only include changes with these tags.
exclude_tags: Exclude changes with these tags.
Example:
```python
reporter.to_text("remediation.txt", style="merged")
```
"""
changes = self.get_all_changes(
include_tags=include_tags,
exclude_tags=exclude_tags,
)
lines = [child.cisco_style_text(style=style) for child in changes]
output_path = Path(file_path)
output_path.write_text("\n".join(lines), encoding="utf-8")
Driver System
hier_config.platforms.driver_base.HConfigDriverBase
Bases: ABC
Defines all hier_config options, rules, and rule checking methods. Override methods as needed.
Source code in hier_config/platforms/driver_base.py
class HConfigDriverBase(ABC):
"""Defines all hier_config options, rules, and rule checking methods.
Override methods as needed.
"""
def __init__(self) -> None:
self.rules = self._instantiate_rules()
def idempotent_for(
self,
config: HConfigChild,
other_children: Iterable[HConfigChild],
) -> HConfigChild | None:
for rule in self.rules.idempotent_commands:
if not config.is_lineage_match(rule.match_rules):
continue
config_key = self._idempotency_key(config, rule.match_rules)
for other_child in other_children:
if not other_child.is_lineage_match(rule.match_rules):
continue
if self._idempotency_key(other_child, rule.match_rules) == config_key:
return other_child
return None
def negate_with(self, config: HConfigChild) -> str | None:
for with_rule in self.rules.negate_with:
if config.is_lineage_match(with_rule.match_rules):
return with_rule.use
return None
def sectional_exit(self, config: HConfigChild) -> str | None:
for exit_rule in self.rules.sectional_exiting:
if config.is_lineage_match(exit_rule.match_rules):
if exit_text := exit_rule.exit_text:
return exit_text
return None
if config.children:
return "exit"
return None
def swap_negation(self, child: HConfigChild) -> HConfigChild:
"""Swap negation of a `child.text`."""
if child.text.startswith(self.negation_prefix):
child.text = child.text_without_negation
else:
child.text = f"{self.negation_prefix}{child.text}"
return child
def _idempotency_key(
self,
config: HConfigChild,
match_rules: tuple[MatchRule, ...],
) -> tuple[str, ...]:
"""Build a structural identity for `config` that respects driver rules.
Args:
config: The child being evaluated for idempotency.
match_rules: The match rules describing the lineage signature.
Returns:
A tuple of string fragments representing the idempotency key.
"""
lineage = tuple(config.lineage())
if len(lineage) != len(match_rules):
return ()
components: list[str] = []
for child, rule in zip(lineage, match_rules, strict=False):
components.append(self._idempotency_component_key(child, rule))
return tuple(components)
def _idempotency_component_key(
self,
child: HConfigChild,
rule: MatchRule,
) -> str:
"""Derive the structural key for a single lineage component.
Args:
child: The lineage child contributing to the key.
rule: The rule governing how to match the child.
Returns:
A string fragment representing the component key.
"""
text = child.text
normalized_text = text.removeprefix(self.negation_prefix)
parts: list[str] = []
parts.extend(self._key_from_equals(rule.equals, text))
parts.extend(self._key_from_prefix(rule.startswith, normalized_text))
parts.extend(self._key_from_suffix(rule.endswith, normalized_text))
parts.extend(self._key_from_contains(rule.contains, normalized_text))
parts.extend(self._key_from_regex(rule.re_search, normalized_text, text))
if not parts:
parts.append(f"text|{normalized_text}")
return ";".join(parts)
@staticmethod
def _key_from_equals(equals: str | frozenset[str] | None, text: str) -> list[str]:
"""Return key fragments constrained by `equals` match rules.
Args:
equals: The equals constraint specified by the rule.
text: The original command text to fall back on for sets.
Returns:
A list containing zero or one key fragments.
"""
if equals is None:
return []
if isinstance(equals, str):
return [f"equals|{equals}"]
return [f"equals|{text}"]
def _key_from_prefix(
self,
prefix: str | tuple[str, ...] | None,
normalized_text: str,
) -> list[str]:
"""Return key fragments for `startswith` match rules.
Args:
prefix: The `startswith` constraint(s) to evaluate.
normalized_text: The command text without the negation prefix.
Returns:
A list containing zero or one key fragments.
"""
if prefix is None:
return []
matched = self._match_prefix(normalized_text, prefix)
if matched is None:
return []
return [f"startswith|{matched}"]
def _key_from_suffix(
self,
suffix: str | tuple[str, ...] | None,
normalized_text: str,
) -> list[str]:
"""Return key fragments for `endswith` match rules.
Args:
suffix: The `endswith` constraint(s) to evaluate.
normalized_text: The command text without the negation prefix.
Returns:
A list containing zero or one key fragments.
"""
if suffix is None:
return []
matched = self._match_suffix(normalized_text, suffix)
if matched is None:
return []
return [f"endswith|{matched}"]
def _key_from_contains(
self,
contains: str | tuple[str, ...] | None,
normalized_text: str,
) -> list[str]:
"""Return key fragments for `contains` match rules.
Args:
contains: The `contains` constraint(s) to evaluate.
normalized_text: The command text without the negation prefix.
Returns:
A list containing zero or one key fragments.
"""
if contains is None:
return []
matched = self._match_contains(normalized_text, contains)
if matched is None:
return []
return [f"contains|{matched}"]
def _key_from_regex(
self,
pattern: str | None,
normalized_text: str,
original_text: str,
) -> list[str]:
"""Return key fragments derived from regex match rules.
Args:
pattern: The regex pattern to match.
normalized_text: The command text without the negation prefix.
original_text: The command text including any negation.
Returns:
A list containing zero or one key fragments.
"""
if pattern is None:
return []
match = search(pattern, normalized_text)
match_source = normalized_text
if match is None:
match = search(pattern, original_text)
match_source = original_text
if match is None:
return []
regex_key = self._normalize_regex_key(pattern, match_source, match)
return [f"re|{regex_key}"]
@staticmethod
def _match_prefix(value: str, prefix: str | tuple[str, ...]) -> str | None:
if isinstance(prefix, tuple):
matches = [candidate for candidate in prefix if value.startswith(candidate)]
if matches:
return max(matches, key=len)
return None
if value.startswith(prefix):
return prefix
return None
@staticmethod
def _match_suffix(value: str, suffix: str | tuple[str, ...]) -> str | None:
if isinstance(suffix, tuple):
matches = [candidate for candidate in suffix if value.endswith(candidate)]
if matches:
return max(matches, key=len)
return None
if value.endswith(suffix):
return suffix
return None
@staticmethod
def _match_contains(value: str, contains: str | tuple[str, ...]) -> str | None:
if isinstance(contains, tuple):
matches = [candidate for candidate in contains if candidate in value]
if matches:
return max(matches, key=len)
return None
if contains in value:
return contains
return None
@staticmethod
def _normalize_regex_key(pattern: str, value: str, match: Match[str]) -> str:
"""Normalize regex matches so equivalent commands hash the same."""
result = match.group(0)
if match.re.groups:
groups = tuple(g or "" for g in match.groups())
if any(groups):
normalized_groups = tuple(group.strip() for group in groups)
if any(normalized_groups):
return "|".join(normalized_groups)
trimmed_pattern = pattern.rstrip("$")
for suffix in (".*", ".+"):
if trimmed_pattern.endswith(suffix):
candidate_pattern = trimmed_pattern[: -len(suffix)]
if not candidate_pattern:
break
trimmed_match = search(candidate_pattern, value)
if trimmed_match is not None:
candidate = trimmed_match.group(0).strip()
if candidate:
return candidate
break
return result.strip()
@property
def declaration_prefix(self) -> str:
return ""
@property
def negation_prefix(self) -> str:
return "no "
@staticmethod
def config_preprocessor(config_text: str) -> str:
return config_text
@staticmethod
@abstractmethod
def _instantiate_rules() -> HConfigDriverRules:
pass
swap_negation(child)
Swap negation of a child.text.
Source code in hier_config/platforms/driver_base.py
def swap_negation(self, child: HConfigChild) -> HConfigChild:
"""Swap negation of a `child.text`."""
if child.text.startswith(self.negation_prefix):
child.text = child.text_without_negation
else:
child.text = f"{self.negation_prefix}{child.text}"
return child
hier_config.platforms.driver_base.HConfigDriverRules
Bases: BaseModel
Pydantic model holding all rule collections for a platform driver.
Each field corresponds to one category of driver behaviour (e.g. negation,
ordering, idempotency). Instantiated by each driver's _instantiate_rules
static method and stored on :class:HConfigDriverBase.
Source code in hier_config/platforms/driver_base.py
class HConfigDriverRules(BaseModel): # pylint: disable=too-many-instance-attributes
"""Pydantic model holding all rule collections for a platform driver.
Each field corresponds to one category of driver behaviour (e.g. negation,
ordering, idempotency). Instantiated by each driver's ``_instantiate_rules``
static method and stored on :class:`HConfigDriverBase`.
"""
full_text_sub: list[FullTextSubRule] = Field(
default_factory=_full_text_sub_rules_default
)
idempotent_commands: list[IdempotentCommandsRule] = Field(
default_factory=_idempotent_commands_rules_default
)
idempotent_commands_avoid: list[IdempotentCommandsAvoidRule] = Field(
default_factory=_idempotent_commands_avoid_rules_default
)
indent_adjust: list[IndentAdjustRule] = Field(
default_factory=_indent_adjust_rules_default
)
indentation: PositiveInt = 2
negation_default_when: list[NegationDefaultWhenRule] = Field(
default_factory=_negation_default_when_rules_default
)
negate_with: list[NegationDefaultWithRule] = Field(
default_factory=_negate_with_rules_default
)
ordering: list[OrderingRule] = Field(default_factory=_ordering_rules_default)
parent_allows_duplicate_child: list[ParentAllowsDuplicateChildRule] = Field(
default_factory=_parent_allows_duplicate_child_rules_default
)
per_line_sub: list[PerLineSubRule] = Field(
default_factory=_per_line_sub_rules_default
)
post_load_callbacks: list[Callable[[HConfig], None]] = Field(
default_factory=_post_load_callbacks_default
)
sectional_exiting: list[SectionalExitingRule] = Field(
default_factory=_sectional_exiting_rules_default
)
sectional_overwrite: list[SectionalOverwriteRule] = Field(
default_factory=_sectional_overwrite_rules_default
)
sectional_overwrite_no_negate: list[SectionalOverwriteNoNegateRule] = Field(
default_factory=_sectional_overwrite_no_negate_rules_default
)
negation_sub: list[NegationSubRule] = Field(
default_factory=_negation_sub_rules_default
)
unused_objects: list[UnusedObjectRule] = Field(
default_factory=_unused_object_rules_default
)
Models
hier_config.models.Platform
Bases: str, Enum
Enumeration of supported network operating system platforms.
Source code in hier_config/models.py
class Platform(str, Enum):
"""Enumeration of supported network operating system platforms."""
ARISTA_EOS = auto()
CISCO_IOS = auto()
CISCO_NXOS = auto()
CISCO_XR = auto()
FORTINET_FORTIOS = auto()
GENERIC = auto() # used in cases where the specific platform is unimportant/unknown
HP_COMWARE5 = auto()
HP_PROCURVE = auto()
HUAWEI_VRP = auto()
JUNIPER_JUNOS = auto()
NOKIA_SRL = auto()
VYOS = auto()
hier_config.models.TextStyle = Literal['without_comments', 'merged', 'with_comments']
module-attribute
hier_config.models.MatchRule
Bases: BaseModel
Flexible predicate for matching an HConfigChild.text value.
All fields are optional; when multiple are set every criterion must match. Used inside all rule types (ordering, idempotency, negation, etc.).
Source code in hier_config/models.py
class MatchRule(BaseModel):
"""Flexible predicate for matching an ``HConfigChild.text`` value.
All fields are optional; when multiple are set every criterion must match.
Used inside all rule types (ordering, idempotency, negation, etc.).
"""
equals: str | frozenset[str] | None = None
startswith: str | tuple[str, ...] | None = None
endswith: str | tuple[str, ...] | None = None
contains: str | tuple[str, ...] | None = None
re_search: str | None = None
hier_config.models.TagRule
Bases: BaseModel
Rule that applies a set of tags to children matching match_rules.
Source code in hier_config/models.py
class TagRule(BaseModel):
"""Rule that applies a set of tags to children matching ``match_rules``."""
match_rules: tuple[MatchRule, ...]
apply_tags: frozenset[str]
hier_config.models.IdempotentCommandsRule
Bases: BaseModel
Rule declaring that a command family is idempotent (last value wins).
Use regex capture groups in MatchRule.re_search to parameterize
idempotency keys — e.g. r"^client (\S+) server-key" makes each
client IP independently idempotent.
Prefer separate rules for unrelated command families rather than
combining them via a tuple startswith in a single MatchRule.
Source code in hier_config/models.py
class IdempotentCommandsRule(BaseModel):
r"""Rule declaring that a command family is idempotent (last value wins).
Use regex capture groups in ``MatchRule.re_search`` to parameterize
idempotency keys — e.g. ``r"^client (\S+) server-key"`` makes each
client IP independently idempotent.
Prefer separate rules for unrelated command families rather than
combining them via a tuple ``startswith`` in a single ``MatchRule``.
"""
match_rules: tuple[MatchRule, ...]
hier_config.models.NegationDefaultWithRule
Bases: BaseModel
Rule replacing negation with a fixed custom command string.
Source code in hier_config/models.py
class NegationDefaultWithRule(BaseModel):
"""Rule replacing negation with a fixed custom command string."""
match_rules: tuple[MatchRule, ...]
use: str
hier_config.models.NegationDefaultWhenRule
Bases: BaseModel
Rule specifying when negation should use the default form.
Source code in hier_config/models.py
class NegationDefaultWhenRule(BaseModel):
"""Rule specifying when negation should use the ``default`` form."""
match_rules: tuple[MatchRule, ...]
hier_config.models.SectionalExitingRule
Bases: BaseModel
Rule defining the exit command for a hierarchical configuration section.
Source code in hier_config/models.py
class SectionalExitingRule(BaseModel):
"""Rule defining the exit command for a hierarchical configuration section."""
match_rules: tuple[MatchRule, ...]
exit_text: str
exit_text_parent_level: bool = False
hier_config.models.SectionalOverwriteRule
Bases: BaseModel
Rule marking a section for full negation + re-creation during remediation.
Source code in hier_config/models.py
class SectionalOverwriteRule(BaseModel):
"""Rule marking a section for full negation + re-creation during remediation."""
match_rules: tuple[MatchRule, ...]
hier_config.models.SectionalOverwriteNoNegateRule
Bases: BaseModel
Rule marking a section for re-creation without prior negation.
Source code in hier_config/models.py
class SectionalOverwriteNoNegateRule(BaseModel):
"""Rule marking a section for re-creation *without* prior negation."""
match_rules: tuple[MatchRule, ...]
hier_config.models.OrderingRule
Bases: BaseModel
Rule assigning an integer weight to commands to control apply order.
Source code in hier_config/models.py
class OrderingRule(BaseModel):
"""Rule assigning an integer weight to commands to control apply order."""
match_rules: tuple[MatchRule, ...]
weight: int
hier_config.models.Dump
Bases: BaseModel
Serialised representation of an entire HConfig tree.
Source code in hier_config/models.py
class Dump(BaseModel):
"""Serialised representation of an entire HConfig tree."""
lines: tuple[DumpLine, ...]
hier_config.models.DumpLine
Bases: BaseModel
A single line from a serialised HConfig tree (see :class:Dump).
Source code in hier_config/models.py
class DumpLine(BaseModel):
"""A single line from a serialised HConfig tree (see :class:`Dump`)."""
depth: NonNegativeInt
text: str
tags: frozenset[str]
comments: frozenset[str]
new_in_config: bool
Utilities
hier_config.utils
hconfig_v2_os_v3_platform_mapper(os_name)
Map a Hier Config v2 operating system name to a v3 Platform enumeration.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
hconfig_v2_os_v3_platform_mapper("CISCO_IOS")
Source code in hier_config/utils.py
def hconfig_v2_os_v3_platform_mapper(os_name: str) -> Platform:
"""Map a Hier Config v2 operating system name to a v3 Platform enumeration.
Args:
os_name (str): The name of the OS as defined in Hier Config v2.
Returns:
Platform: The corresponding Platform enumeration for Hier Config v3.
Example:
>>> hconfig_v2_os_v3_platform_mapper("CISCO_IOS")
<Platform.CISCO_IOS: 'ios'>
"""
return HCONFIG_PLATFORM_V2_TO_V3_MAPPING.get(os_name, Platform.GENERIC)
hconfig_v3_platform_v2_os_mapper(platform)
Map a Hier Config v3 Platform enumeration to a v2 operating system name.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Example
hconfig_v3_platform_v2_os_mapper(Platform.CISCO_IOS) "ios"
Source code in hier_config/utils.py
def hconfig_v3_platform_v2_os_mapper(platform: Platform) -> str:
"""Map a Hier Config v3 Platform enumeration to a v2 operating system name.
Args:
platform (Platform): A Platform enumeration from Hier Config v3.
Returns:
str: The corresponding OS name for Hier Config v2.
Example:
>>> hconfig_v3_platform_v2_os_mapper(Platform.CISCO_IOS)
"ios"
"""
for os_name, plat in HCONFIG_PLATFORM_V2_TO_V3_MAPPING.items():
if plat == platform:
return os_name
return "generic"
load_hconfig_v2_options(v2_options, platform)
Load Hier Config v2 options to v3 driver format from either a dictionary or a file.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Source code in hier_config/utils.py
def load_hconfig_v2_options(
v2_options: dict[str, Any] | str, platform: Platform
) -> HConfigDriverBase:
"""Load Hier Config v2 options to v3 driver format from either a dictionary or a file.
Args:
v2_options (Union[dict, str]): Either a dictionary containing v2 options or
a file path to a YAML file containing the v2 options.
platform (Platform): The Hier Config v3 Platform enum for the target platform.
Returns:
HConfigDriverBase: A v3 driver instance with the migrated rules.
"""
if isinstance(v2_options, str):
v2_options = yaml.safe_load(read_text_from_file(file_path=v2_options))
if not isinstance(v2_options, dict):
msg = "v2_options must be a dictionary or a valid file path."
raise TypeError(msg)
driver = get_hconfig_driver(platform)
# Process simple rules that only need match_rules
simple_rules: tuple[tuple[str, type[Any], Callable[[Any], None]], ...] = (
(
"sectional_overwrite",
SectionalOverwriteRule,
driver.rules.sectional_overwrite.append,
),
(
"sectional_overwrite_no_negate",
SectionalOverwriteNoNegateRule,
driver.rules.sectional_overwrite_no_negate.append,
),
(
"parent_allows_duplicate_child",
ParentAllowsDuplicateChildRule,
driver.rules.parent_allows_duplicate_child.append,
),
(
"idempotent_commands_blacklist",
IdempotentCommandsAvoidRule,
driver.rules.idempotent_commands_avoid.append,
),
(
"idempotent_commands",
IdempotentCommandsRule,
driver.rules.idempotent_commands.append,
),
(
"negation_default_when",
NegationDefaultWhenRule,
driver.rules.negation_default_when.append,
),
)
for key, rule_class, append_to in simple_rules:
_process_simple_rules(v2_options, key, rule_class, append_to)
# Process rules that require custom handling
_process_custom_rules(v2_options, driver)
return driver
load_hconfig_v2_options_from_file(options_file, platform)
Load Hier Config v2 options file to v3 driver format.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Source code in hier_config/utils.py
def load_hconfig_v2_options_from_file(
options_file: str, platform: Platform
) -> HConfigDriverBase:
"""Load Hier Config v2 options file to v3 driver format.
Args:
options_file (str): The v2 options file.
platform (Platform): The Hier Config v3 Platform enum for the target platform.
Returns:
HConfigDriverBase: A v3 driver instance with the migrated rules.
"""
hconfig_options = yaml.safe_load(read_text_from_file(file_path=options_file))
return load_hconfig_v2_options(v2_options=hconfig_options, platform=platform)
load_hconfig_v2_tags(v2_tags)
Convert v2-style tags into v3-style TagRule Pydantic objects for Hier Config.
| Parameters: |
|
|---|
| Returns: |
|---|
Source code in hier_config/utils.py
def load_hconfig_v2_tags(
v2_tags: list[dict[str, Any]] | str,
) -> tuple["TagRule"] | tuple["TagRule", ...]:
"""Convert v2-style tags into v3-style TagRule Pydantic objects for Hier Config.
Args:
v2_tags (Union[list[dict[str, Any]], str]):
Either a list of dictionaries representing v2-style tags or a file path
to a YAML file containing the v2-style tags.
- If a list is provided, each dictionary should contain:
- `lineage`: A list of dictionaries with rules (e.g., `startswith`, `endswith`).
- `add_tags`: A string representing the tag to add.
- If a file path is provided, it will be read and parsed as YAML.
Returns:
Tuple[TagRule]: A tuple of TagRule Pydantic objects representing v3-style tags.
"""
# Load tags from a file if a string is provided
if isinstance(v2_tags, str):
v2_tags = yaml.safe_load(read_text_from_file(file_path=v2_tags))
# Ensure v2_tags is a list
if not isinstance(v2_tags, list):
msg = "v2_tags must be a list of dictionaries or a valid file path."
raise TypeError(msg)
v3_tags: list[TagRule] = []
for v2_tag in v2_tags:
if "lineage" in v2_tag and "add_tags" in v2_tag:
# Extract the v2 fields
lineage_rules = v2_tag["lineage"]
tags = v2_tag["add_tags"]
# Convert to MatchRule objects
match_rules = tuple(
match_rule
for lineage in lineage_rules
if (match_rule := _set_match_rule(lineage)) is not None
)
# Create the TagRule object
v3_tag = TagRule(match_rules=match_rules, apply_tags=frozenset([tags]))
v3_tags.append(v3_tag)
return tuple(v3_tags)
load_hier_config_tags(tags_file)
Loads and validates Hier Config tags from a YAML file.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Source code in hier_config/utils.py
def load_hier_config_tags(tags_file: str) -> tuple[TagRule, ...]:
"""Loads and validates Hier Config tags from a YAML file.
Args:
tags_file (str): Path to the YAML file containing the tags.
Returns:
Tuple[TagRule, ...]: A tuple of validated TagRule objects.
"""
tags_data = yaml.safe_load(read_text_from_file(file_path=tags_file))
return TypeAdapter(tuple[TagRule, ...]).validate_python(tags_data)
read_text_from_file(file_path)
Function that loads the contents of a file into memory.
| Parameters: |
|
|---|
| Returns: |
|
|---|
Source code in hier_config/utils.py
def read_text_from_file(file_path: str) -> str:
"""Function that loads the contents of a file into memory.
Args:
file_path (str): The path to the configuration file.
Returns:
str: The configuration file contents as a string.
"""
return Path(file_path).read_text(encoding="utf-8")