Loading edge_cloud_management_api/controllers/intent_controller.py 0 → 100644 +55 −0 Original line number Diff line number Diff line from __future__ import annotations from flask import jsonify from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.intent_service import SRMIntentClientFactory _srm_factory = SRMIntentClientFactory() _srm_client = _srm_factory.create_client() def _error(status: int, code: str, message: str): return jsonify({"code": code, "reason": message, "status": str(status)}), status def list_intent( fields: str | None = None, offset: int | None = None, limit: int | None = None, ): try: intents = _srm_client.list_intents(fields=fields, offset=offset, limit=limit) response = jsonify(intents) response.headers["X-Result-Count"] = str(len(intents)) return response, 200 except Exception as e: logger.exception("Unexpected error in list_intent: %s", e) return _error(500, "INTERNAL_ERROR", f"An error occurred: {str(e)}") def create_intent(body: dict, fields: str | None = None): try: srm_response = _srm_client.create_intent(body) if isinstance(srm_response, dict) and "error" in srm_response: status = srm_response.get("status_code", 500) return _error(status, "SRM_ERROR", srm_response["error"]) return jsonify(srm_response), 201 except Exception as e: logger.exception("Unexpected error in create_intent: %s", e) return _error(500, "INTERNAL_ERROR", f"An error occurred: {str(e)}") def delete_intent_by_id(id: str): # noqa: A002 try: result = _srm_client.delete_intent(id) if isinstance(result, dict) and "error" in result: status = result.get("status_code", 500) return _error(status, "SRM_ERROR", result["error"]) return "", 204 except Exception as e: logger.exception("Unexpected error in delete_intent_by_id: %s", e) return _error(500, "INTERNAL_ERROR", f"An error occurred: {str(e)}") Loading
edge_cloud_management_api/controllers/intent_controller.py 0 → 100644 +55 −0 Original line number Diff line number Diff line from __future__ import annotations from flask import jsonify from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.intent_service import SRMIntentClientFactory _srm_factory = SRMIntentClientFactory() _srm_client = _srm_factory.create_client() def _error(status: int, code: str, message: str): return jsonify({"code": code, "reason": message, "status": str(status)}), status def list_intent( fields: str | None = None, offset: int | None = None, limit: int | None = None, ): try: intents = _srm_client.list_intents(fields=fields, offset=offset, limit=limit) response = jsonify(intents) response.headers["X-Result-Count"] = str(len(intents)) return response, 200 except Exception as e: logger.exception("Unexpected error in list_intent: %s", e) return _error(500, "INTERNAL_ERROR", f"An error occurred: {str(e)}") def create_intent(body: dict, fields: str | None = None): try: srm_response = _srm_client.create_intent(body) if isinstance(srm_response, dict) and "error" in srm_response: status = srm_response.get("status_code", 500) return _error(status, "SRM_ERROR", srm_response["error"]) return jsonify(srm_response), 201 except Exception as e: logger.exception("Unexpected error in create_intent: %s", e) return _error(500, "INTERNAL_ERROR", f"An error occurred: {str(e)}") def delete_intent_by_id(id: str): # noqa: A002 try: result = _srm_client.delete_intent(id) if isinstance(result, dict) and "error" in result: status = result.get("status_code", 500) return _error(status, "SRM_ERROR", result["error"]) return "", 204 except Exception as e: logger.exception("Unexpected error in delete_intent_by_id: %s", e) return _error(500, "INTERNAL_ERROR", f"An error occurred: {str(e)}")