Skip to content

platform_service

platform_service

Service layer for platform information and batch operations.

PlatformService

Service for handling platform-related operations.

Source code in hier_config_api/services/platform_service.py 
 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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class PlatformService:
    """Service for handling platform-related operations."""

    # Common platform definitions
    PLATFORMS = {
        "cisco_ios": PlatformInfo(
            platform_name="cisco_ios",
            display_name="Cisco IOS",
            vendor="Cisco",
            supported=True,
        ),
        "cisco_nxos": PlatformInfo(
            platform_name="cisco_nxos",
            display_name="Cisco NX-OS",
            vendor="Cisco",
            supported=True,
        ),
        "cisco_iosxr": PlatformInfo(
            platform_name="cisco_iosxr",
            display_name="Cisco IOS-XR",
            vendor="Cisco",
            supported=True,
        ),
        "juniper_junos": PlatformInfo(
            platform_name="juniper_junos",
            display_name="Juniper Junos",
            vendor="Juniper",
            supported=True,
        ),
        "arista_eos": PlatformInfo(
            platform_name="arista_eos",
            display_name="Arista EOS",
            vendor="Arista",
            supported=True,
        ),
    }

    @staticmethod
    def list_platforms() -> list[PlatformInfo]:
        """List all supported platforms."""
        return list(PlatformService.PLATFORMS.values())

    @staticmethod
    def get_platform_rules(platform: str) -> PlatformRules:
        """Get platform-specific rules."""
        # This is a simplified version
        # In a real implementation, you'd load this from hier_config's driver options
        return PlatformRules(
            platform_name=platform,
            negation_default_when=[],
            negation_negate_with=[],
            ordering=[],
            idempotent_commands_avoid=[],
            idempotent_commands=[],
        )

    @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 validate_config(platform: str, config_text: str) -> dict[str, Any]:
        """Validate configuration for a platform."""
        warnings = []
        errors = []
        is_valid = True

        try:
            # Try to parse the configuration
            platform_enum = PlatformService._get_platform(platform)
            get_hconfig(platform_enum, config_text)

            # Basic validation checks
            if not config_text.strip():
                warnings.append("Configuration is empty")
                is_valid = False

            # Platform-specific validation could be added here
            if platform == "cisco_ios":
                # Check for common Cisco IOS patterns
                if "hostname" not in config_text:
                    warnings.append("No hostname configured")

        except Exception as e:
            errors.append(f"Configuration parsing error: {str(e)}")
            is_valid = False

        return {
            "platform": platform,
            "is_valid": is_valid,
            "warnings": warnings,
            "errors": errors,
        }

    @staticmethod
    def create_batch_job(device_configs: list[dict[str, Any]]) -> dict[str, Any]:
        """Create a batch remediation job."""
        return {
            "status": "pending",
            "progress": 0.0,
            "total_devices": len(device_configs),
            "completed_devices": 0,
            "failed_devices": 0,
            "device_configs": device_configs,
            "results": [],
        }

    @staticmethod
    def process_batch_job(job_data: dict[str, Any]) -> dict[str, Any]:
        """Process a batch job (simplified synchronous version)."""
        results = []
        completed = 0
        failed = 0

        for device_config in job_data["device_configs"]:
            try:
                # Process each device
                platform = device_config.get("platform", "cisco_ios")
                running_config = device_config.get("running_config", "")
                intended_config = device_config.get("intended_config", "")

                platform_enum = PlatformService._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

                results.append(
                    {
                        "device_id": device_config.get("device_id"),
                        "status": "success",
                        "remediation": str(remediation) if remediation else "",
                        "rollback": str(rollback) if rollback else "",
                    }
                )
                completed += 1

            except Exception as e:
                results.append(
                    {
                        "device_id": device_config.get("device_id"),
                        "status": "failed",
                        "error": str(e),
                    }
                )
                failed += 1

        # Update job data
        job_data.update(
            {
                "status": "completed",
                "progress": 100.0,
                "completed_devices": completed,
                "failed_devices": failed,
                "results": results,
            }
        )

        return job_data

list_platforms() staticmethod

List all supported platforms.

Source code in hier_config_api/services/platform_service.py 
47
48
49
50
@staticmethod
def list_platforms() -> list[PlatformInfo]:
    """List all supported platforms."""
    return list(PlatformService.PLATFORMS.values())

get_platform_rules(platform) staticmethod

Get platform-specific rules.

Source code in hier_config_api/services/platform_service.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
@staticmethod
def get_platform_rules(platform: str) -> PlatformRules:
    """Get platform-specific rules."""
    # This is a simplified version
    # In a real implementation, you'd load this from hier_config's driver options
    return PlatformRules(
        platform_name=platform,
        negation_default_when=[],
        negation_negate_with=[],
        ordering=[],
        idempotent_commands_avoid=[],
        idempotent_commands=[],
    )

validate_config(platform, config_text) staticmethod

Validate configuration for a platform.

Source code in hier_config_api/services/platform_service.py 
 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
@staticmethod
def validate_config(platform: str, config_text: str) -> dict[str, Any]:
    """Validate configuration for a platform."""
    warnings = []
    errors = []
    is_valid = True

    try:
        # Try to parse the configuration
        platform_enum = PlatformService._get_platform(platform)
        get_hconfig(platform_enum, config_text)

        # Basic validation checks
        if not config_text.strip():
            warnings.append("Configuration is empty")
            is_valid = False

        # Platform-specific validation could be added here
        if platform == "cisco_ios":
            # Check for common Cisco IOS patterns
            if "hostname" not in config_text:
                warnings.append("No hostname configured")

    except Exception as e:
        errors.append(f"Configuration parsing error: {str(e)}")
        is_valid = False

    return {
        "platform": platform,
        "is_valid": is_valid,
        "warnings": warnings,
        "errors": errors,
    }

create_batch_job(device_configs) staticmethod

Create a batch remediation job.

Source code in hier_config_api/services/platform_service.py 
113
114
115
116
117
118
119
120
121
122
123
124
@staticmethod
def create_batch_job(device_configs: list[dict[str, Any]]) -> dict[str, Any]:
    """Create a batch remediation job."""
    return {
        "status": "pending",
        "progress": 0.0,
        "total_devices": len(device_configs),
        "completed_devices": 0,
        "failed_devices": 0,
        "device_configs": device_configs,
        "results": [],
    }

process_batch_job(job_data) staticmethod

Process a batch job (simplified synchronous version).

Source code in hier_config_api/services/platform_service.py 
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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@staticmethod
def process_batch_job(job_data: dict[str, Any]) -> dict[str, Any]:
    """Process a batch job (simplified synchronous version)."""
    results = []
    completed = 0
    failed = 0

    for device_config in job_data["device_configs"]:
        try:
            # Process each device
            platform = device_config.get("platform", "cisco_ios")
            running_config = device_config.get("running_config", "")
            intended_config = device_config.get("intended_config", "")

            platform_enum = PlatformService._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

            results.append(
                {
                    "device_id": device_config.get("device_id"),
                    "status": "success",
                    "remediation": str(remediation) if remediation else "",
                    "rollback": str(rollback) if rollback else "",
                }
            )
            completed += 1

        except Exception as e:
            results.append(
                {
                    "device_id": device_config.get("device_id"),
                    "status": "failed",
                    "error": str(e),
                }
            )
            failed += 1

    # Update job data
    job_data.update(
        {
            "status": "completed",
            "progress": 100.0,
            "completed_devices": completed,
            "failed_devices": failed,
            "results": results,
        }
    )

    return job_data