Edit intent_service.py

import requests

from requests.exceptions import Timeout, ConnectionError

from edge_cloud_management_api.configs.env_config import config

from edge_cloud_management_api.managers.log_manager import logger

class SRMIntentClient:

    """

    Client for SRM Intent API endpoints.

    Uses the same base URL and credentials as PiEdgeAPIClient.

    """

    def __init__(self, base_url: str, username: str, password: str):

        self.base_url = base_url

        self.username = username

        self.password = password

    def _get_headers(self) -> dict:

        return {"Content-Type": "application/json"}

    def create_intent(self, body: dict) -> dict:

         url = f"{self.base_url}/intent"

         try:

             response = requests.post(url, json=body, headers=self._get_headers(), verify=False)

             response.raise_for_status()

             return response.json()

         except Timeout:

             return {"error": "The request to SRM timed out."}

         except ConnectionError:

             return {"error": "Failed to connect to SRM."}

         except requests.exceptions.HTTPError as http_err:

             return {"error": f"HTTP error occurred: {http_err}", "status_code": response.status_code}

         except Exception as err:

             return {"error": f"An unexpected error occurred: {err}"}

    def list_intents(

        self,

        fields: str | None = None,

        offset: int | None = None,

        limit: int | None = None,

    ) -> list:

         url = f"{self.base_url}/intent"

         params = {k: v for k, v in {"fields": fields, "offset": offset, "limit": limit}.items() if v is not None}

         try:

             response = requests.get(url, params=params, headers=self._get_headers(), verify=False)

             response.raise_for_status()

             return response.json()

         except Timeout:

             return []

         except ConnectionError:

             return []

         except requests.exceptions.HTTPError as http_err:

             logger.error(f"HTTP error occurred: {http_err}")

             return []

         except Exception as err:

             logger.error(f"An unexpected error occurred: {err}")

             return []

    def delete_intent(self, intent_id: str) -> bool:

         url = f"{self.base_url}/intent/{intent_id}"

         try:

             response = requests.delete(url, headers=self._get_headers(), verify=False)

             response.raise_for_status()

             return True

         except Timeout:

             return {"error": "The request to SRM timed out."}

         except ConnectionError:

             return {"error": "Failed to connect to SRM."}

         except requests.exceptions.HTTPError as http_err:

             return {"error": f"HTTP error occurred: {http_err}", "status_code": response.status_code}

         except Exception as err:

             return {"error": f"An unexpected error occurred: {err}"}

class SRMIntentClientFactory:

    """

    Factory for SRMIntentClient.

    Reuses the same SRM credentials already configured for PiEdgeAPIClient.

    """

    def __init__(self):

        self.base_url = config.SRM_HOST

        self.username = config.PI_EDGE_USERNAME

        self.password = config.PI_EDGE_PASSWORD

    def create_client(self) -> SRMIntentClient:

        return SRMIntentClient(

            base_url=self.base_url,

            username=self.username,

            password=self.password,

        )

 No newline at end of file