Skip to content

config_service

config_service

Service layer for configuration operations.

ConfigService

Service for handling configuration operations.

Source code in hier_config_api/services/config_service.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class ConfigService:
    """Service for handling configuration operations."""

    @staticmethod
    def _get_platform(platform_str: str) -> Platform:
        """Convert platform string to Platform enum."""
        platform_map = {
            "cisco_ios": Platform.CISCO_IOS,
            "cisco_nxos": Platform.CISCO_NXOS,
            "cisco_iosxr": Platform.CISCO_XR,
            "juniper_junos": Platform.JUNIPER_JUNOS,
            "arista_eos": Platform.ARISTA_EOS,
            "generic": Platform.GENERIC,
        }
        return platform_map.get(platform_str.lower(), Platform.GENERIC)

    @staticmethod
    def parse_config(platform: str, config_text: str) -> dict[str, Any]:
        """Parse configuration text into structured format."""
        platform_enum = ConfigService._get_platform(platform)
        hconfig = get_hconfig(platform_enum, config_text)

        # Convert HConfig tree to dictionary representation
        def config_to_dict(config_obj: Any) -> dict[str, Any]:
            result: dict[str, Any] = {
                "text": str(config_obj),
                "children": [],
            }
            if hasattr(config_obj, "children"):
                for child in config_obj.children:
                    result["children"].append(config_to_dict(child))
            return result

        return config_to_dict(hconfig)

    @staticmethod
    def compare_configs(
        platform: str, running_config: str, intended_config: str
    ) -> tuple[str, bool]:
        """Compare two configurations and return unified diff."""
        platform_enum = ConfigService._get_platform(platform)
        running_hconfig = get_hconfig(platform_enum, running_config)
        intended_hconfig = get_hconfig(platform_enum, intended_config)

        workflow = WorkflowRemediation(running_hconfig, intended_hconfig)
        remediation = workflow.remediation_config
        rollback = workflow.rollback_config

        diff_lines = []

        # Generate unified diff format
        if remediation:
            diff_lines.append("--- running_config")
            diff_lines.append("+++ intended_config")
            for line in str(remediation).splitlines():
                if line.strip():
                    diff_lines.append(f"+ {line}")

        if rollback:
            for line in str(rollback).splitlines():
                if line.strip():
                    diff_lines.append(f"- {line}")

        unified_diff = "\n".join(diff_lines) if diff_lines else "No differences found"
        has_changes = bool(diff_lines)

        return unified_diff, has_changes

    @staticmethod
    def predict_config(platform: str, current_config: str, commands_to_apply: str) -> str:
        """Predict configuration state after applying commands."""
        # Simple merge: append new commands
        config_lines = current_config.splitlines()
        command_lines = commands_to_apply.splitlines()
        predicted_lines = config_lines + command_lines

        return "\n".join(predicted_lines)

    @staticmethod
    def merge_configs(platform: str, configs: list[str]) -> str:
        """Merge multiple configurations into one."""
        if not configs:
            return ""

        if len(configs) == 1:
            return configs[0]

        platform_enum = ConfigService._get_platform(platform)

        # Use the first config as base
        merged = configs[0]

        # Merge each subsequent config
        for config in configs[1:]:
            running_hconfig = get_hconfig(platform_enum, merged)
            intended_hconfig = get_hconfig(platform_enum, config)

            workflow = WorkflowRemediation(running_hconfig, intended_hconfig)
            remediation = workflow.remediation_config
            if remediation:
                merged += "\n" + str(remediation)

        return merged

    @staticmethod
    def search_config(
        platform: str,
        config_text: str,
        equals: str | None = None,
        contains: str | None = None,
        startswith: str | None = None,
        regex_pattern: str | None = None,
    ) -> list[str]:
        """Search configuration for matching lines."""
        platform_enum = ConfigService._get_platform(platform)
        hconfig = get_hconfig(platform_enum, config_text)

        matches = []

        def search_recursive(config_obj: Any) -> None:
            config_line = str(config_obj).strip()

            # Check matching conditions
            is_match = False
            if equals and config_line == equals:
                is_match = True
            elif contains and contains in config_line:
                is_match = True
            elif startswith and config_line.startswith(startswith):
                is_match = True
            elif regex_pattern and re.search(regex_pattern, config_line):
                is_match = True

            if is_match:
                matches.append(config_line)

            # Recursively search children
            if hasattr(config_obj, "children"):
                for child in config_obj.children:
                    search_recursive(child)

        search_recursive(hconfig)
        return matches

parse_config(platform, config_text) staticmethod

Parse configuration text into structured format.

Source code in hier_config_api/services/config_service.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@staticmethod
def parse_config(platform: str, config_text: str) -> dict[str, Any]:
    """Parse configuration text into structured format."""
    platform_enum = ConfigService._get_platform(platform)
    hconfig = get_hconfig(platform_enum, config_text)

    # Convert HConfig tree to dictionary representation
    def config_to_dict(config_obj: Any) -> dict[str, Any]:
        result: dict[str, Any] = {
            "text": str(config_obj),
            "children": [],
        }
        if hasattr(config_obj, "children"):
            for child in config_obj.children:
                result["children"].append(config_to_dict(child))
        return result

    return config_to_dict(hconfig)

compare_configs(platform, running_config, intended_config) staticmethod

Compare two configurations and return unified diff.

Source code in hier_config_api/services/config_service.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@staticmethod
def compare_configs(
    platform: str, running_config: str, intended_config: str
) -> tuple[str, bool]:
    """Compare two configurations and return unified diff."""
    platform_enum = ConfigService._get_platform(platform)
    running_hconfig = get_hconfig(platform_enum, running_config)
    intended_hconfig = get_hconfig(platform_enum, intended_config)

    workflow = WorkflowRemediation(running_hconfig, intended_hconfig)
    remediation = workflow.remediation_config
    rollback = workflow.rollback_config

    diff_lines = []

    # Generate unified diff format
    if remediation:
        diff_lines.append("--- running_config")
        diff_lines.append("+++ intended_config")
        for line in str(remediation).splitlines():
            if line.strip():
                diff_lines.append(f"+ {line}")

    if rollback:
        for line in str(rollback).splitlines():
            if line.strip():
                diff_lines.append(f"- {line}")

    unified_diff = "\n".join(diff_lines) if diff_lines else "No differences found"
    has_changes = bool(diff_lines)

    return unified_diff, has_changes

predict_config(platform, current_config, commands_to_apply) staticmethod

Predict configuration state after applying commands.

Source code in hier_config_api/services/config_service.py 
77
78
79
80
81
82
83
84
85
@staticmethod
def predict_config(platform: str, current_config: str, commands_to_apply: str) -> str:
    """Predict configuration state after applying commands."""
    # Simple merge: append new commands
    config_lines = current_config.splitlines()
    command_lines = commands_to_apply.splitlines()
    predicted_lines = config_lines + command_lines

    return "\n".join(predicted_lines)

merge_configs(platform, configs) staticmethod

Merge multiple configurations into one.

Source code in hier_config_api/services/config_service.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@staticmethod
def merge_configs(platform: str, configs: list[str]) -> str:
    """Merge multiple configurations into one."""
    if not configs:
        return ""

    if len(configs) == 1:
        return configs[0]

    platform_enum = ConfigService._get_platform(platform)

    # Use the first config as base
    merged = configs[0]

    # Merge each subsequent config
    for config in configs[1:]:
        running_hconfig = get_hconfig(platform_enum, merged)
        intended_hconfig = get_hconfig(platform_enum, config)

        workflow = WorkflowRemediation(running_hconfig, intended_hconfig)
        remediation = workflow.remediation_config
        if remediation:
            merged += "\n" + str(remediation)

    return merged

search_config(platform, config_text, equals=None, contains=None, startswith=None, regex_pattern=None) staticmethod

Search configuration for matching lines.

Source code in hier_config_api/services/config_service.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@staticmethod
def search_config(
    platform: str,
    config_text: str,
    equals: str | None = None,
    contains: str | None = None,
    startswith: str | None = None,
    regex_pattern: str | None = None,
) -> list[str]:
    """Search configuration for matching lines."""
    platform_enum = ConfigService._get_platform(platform)
    hconfig = get_hconfig(platform_enum, config_text)

    matches = []

    def search_recursive(config_obj: Any) -> None:
        config_line = str(config_obj).strip()

        # Check matching conditions
        is_match = False
        if equals and config_line == equals:
            is_match = True
        elif contains and contains in config_line:
            is_match = True
        elif startswith and config_line.startswith(startswith):
            is_match = True
        elif regex_pattern and re.search(regex_pattern, config_line):
            is_match = True

        if is_match:
            matches.append(config_line)

        # Recursively search children
        if hasattr(config_obj, "children"):
            for child in config_obj.children:
                search_recursive(child)

    search_recursive(hconfig)
    return matches