SRM Refactor: FastAPI with new Domain Model and PostgreSQL adapter (!14) · Merge requests · OOP / code / service-resource-manager

service-resource-manager-implementation/src/controllers/init.py

deleted100644 → 0

+0 −2

Changes for service-resource-manager-implementation/src/controllers/__init__.py: 0 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		import urllib3
		urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
		No newline at end of file

service-resource-manager-implementation/src/controllers/app_instance_controller.py

deleted100644 → 0

+0 −65

Changes for service-resource-manager-implementation/src/controllers/app_instance_controller.py: 0 added lines, 65 removed lines.
Original line number	Diff line number	Diff line
		import logging

		logger = logging.getLogger(__name__)
		logging.basicConfig(level=logging.DEBUG)

		def deploy_app(helm_request_body):
		"""
		Deploys an application via Helm.
		Adapter-level function:
		- accepts Python dict
		- returns Python dict
		- raises exceptions on failure
		"""

		logger.info("Submitting request to edge cloud transformation function")

		if not isinstance(helm_request_body, dict):
		raise ValueError("Invalid request body")

		if not helm_request_body.get("uri"):
		raise ValueError("Helm chart uri is missing")

		if not helm_request_body.get("deployment_name"):
		raise ValueError("Helm chart deployment name is missing")

		try:
		# submit_helm_chart MUST return dict or list
		result = submit_helm_chart(helm_request_body)

		if not isinstance(result, (dict, list)):
		raise TypeError(
		f"submit_helm_chart returned unsupported type: {type(result)}"
		)

		return result

		except Exception as e:
		logger.exception("Error submitting helm chart")
		raise


		def get_app_instances(body):
		"""
		Instantiates app on the edge cloud platform.
		Adapter-level function.
		"""

		logger.info("Deploying app instance")

		if not isinstance(body, dict):
		raise ValueError("Invalid request body")

		try:
		result = app_deploy(body)

		if not isinstance(result, (dict, list)):
		raise TypeError(
		f"app_deploy returned unsupported type: {type(result)}"
		)

		return result

		except Exception as e:
		logger.exception("Error instantiating app")
		raise

service-resource-manager-implementation/src/controllers/artifact_controller.py

deleted100644 → 0

+0 −38

Changes for service-resource-manager-implementation/src/controllers/artifact_controller.py: 0 added lines, 38 removed lines.
Original line number	Diff line number	Diff line
		import logging
		import json
		import connexion
		from src.models.copy_artifact_model import CopyArtifactModel
		from src.models.artifact_exists_model import ArtifactExistsModel
		from src.utils import artifact_connector

		def artifact_exists():
		if connexion.request.is_json:
		try:
		# artifact = ArtifactExistsModel.from_dict(connexion.request.get_json())
		if connexion.request.get_json().get('registry_url') is None:
		return 'The registyr url must be provided', 400
		if connexion.request.get_json().get('artefact_name') is None:
		return 'Image name must be provided', 400
		response = artifact_connector.artifact_exists(connexion.request.get_json())
		return response.json()
		except Exception as e:
		logging.error(e.args)
		return e.args, 500
		else:
		return 'Request format must be JSON', 400

		def copy_artifact():
		if connexion.request.is_json:
		try:
		artifact = CopyArtifactModel.from_dict(connexion.request.get_json())
		if artifact.dst_registry is None or artifact.src_registry is None:
		return 'The source and destination registries must be provided', 400
		if artifact.src_image_name is None:
		return 'The image name must be provided', 400
		response = artifact_connector.copy_artifact(artifact.to_dict())
		return response.json()
		except Exception as e:
		logging.error(e.args)
		return e.args
		else:
		return 'Request format must be JSON', 400
		No newline at end of file

service-resource-manager-implementation/src/controllers/edge_cloud_management_controller.py

deleted100644 → 0

+0 −216

Changes for service-resource-manager-implementation/src/controllers/edge_cloud_management_controller.py: 0 added lines, 216 removed lines.
Original line number	Diff line number	Diff line
		import connexion
		import logging
		import uuid
		from os import environ
		from sunrise6g_opensdk import Sdk as sdkclient

		logger = logging.getLogger(__name__)
		logging.basicConfig(level=logging.DEBUG)

		# ---------------------------------------------------------------------
		# Adapter initialization
		# ---------------------------------------------------------------------

		edgecloud_adapter = None

		if environ.get("EDGE_CLOUD_ADAPTER_NAME"):
		edgecloud_adapter_name = environ["EDGE_CLOUD_ADAPTER_NAME"]
		edgecloud_adapter_base_url = environ.get("ADAPTER_BASE_URL")

		edgecloud_adapter_specs = {
		"client_name": edgecloud_adapter_name,
		"base_url": edgecloud_adapter_base_url,
		}
		edgecloud_adapter_specs.update(environ)

		logger.info("Creating edge cloud adapter with env: %s", edgecloud_adapter_specs)

		adapters = sdkclient.create_adapters_from(
		adapter_specs={"edgecloud": edgecloud_adapter_specs}
		)
		edgecloud_adapter = adapters.get("edgecloud")

		if edgecloud_adapter is None:
		raise RuntimeError("Edge cloud adapter could not be initialized")

		# ---------------------------------------------------------------------
		# Helpers: normalize adapter responses
		# ---------------------------------------------------------------------

		def _safe_http_json_response(response):
		"""
		Normalize adapter responses:
		- list / dict → return directly (200)
		- requests.Response → parse JSON safely
		"""

		# Adapter already returned Python data
		if isinstance(response, (list, dict)):
		return response, 200

		# Adapter returned nothing
		if response is None:
		return {"error": "Adapter returned no response"}, 502

		# Adapter returned a bare status code integer
		if isinstance(response, int):
		return {}, response

		# Adapter returned a plain string message
		if isinstance(response, str):
		return {"message": response}, 200

		# Adapter returned HTTP response
		try:
		if not response.content:
		return {}, response.status_code
		return response.json(), response.status_code
		except Exception as e:
		logger.exception("Failed to parse adapter HTTP response")
		return {
		"error": "Invalid response from adapter",
		"details": str(e)
		}, 502


		def _ensure_edge_cloud_region(data):
		"""
		Ensure every zone/node object carries an edgeCloudRegion field.
		Falls back to 'unknown' when the adapter omits the field.
		"""
		if isinstance(data, list):
		for item in data:
		if isinstance(item, dict) and not item.get("edgeCloudRegion"):
		item["edgeCloudRegion"] = "unknown"
		elif isinstance(data, dict):
		if not data.get("edgeCloudRegion"):
		data["edgeCloudRegion"] = "unknown"
		return data

		# ---------------------------------------------------------------------
		# Catalogue (onboarded service functions)
		# ---------------------------------------------------------------------

		def register_service_function(body=None):
		if not connexion.request.is_json:
		return {"error": "Invalid JSON payload"}, 400

		try:
		payload = connexion.request.get_json()
		if not payload.get("appId"):
		payload["appId"] = str(uuid.uuid4())
		response = edgecloud_adapter.onboard_app(payload)
		return _safe_http_json_response(response)
		except Exception as e:
		logger.exception("Failed to register service function")
		return {"error": "Registration failed", "details": str(e)}, 500


		def get_service_function(service_function_id: str):
		try:
		response = edgecloud_adapter.get_onboarded_app(service_function_id)
		return _safe_http_json_response(response)
		except Exception as e:
		logger.exception("Failed to get service function")
		return {"error": "Fetch failed", "details": str(e)}, 500


		def get_service_functions():
		try:
		response = edgecloud_adapter.get_all_onboarded_apps()
		return _safe_http_json_response(response)
		except Exception as e:
		logger.exception("Failed to list service functions")
		return {"error": "Fetch failed", "details": str(e)}, 500


		def deregister_service_function(service_function_id: str):
		try:
		response = edgecloud_adapter.delete_onboarded_app(service_function_id)
		return _safe_http_json_response(response)
		except Exception as e:
		logger.exception("Failed to deregister service function")
		return {"error": "Deletion failed", "details": str(e)}, 500

		# ---------------------------------------------------------------------
		# Deployment
		# ---------------------------------------------------------------------

		def deploy_service_function():
		if not connexion.request.is_json:
		return {"error": "Invalid JSON payload"}, 400

		try:
		body = connexion.request.get_json()

		response = edgecloud_adapter.deploy_app(
		body.get("appId"),
		body.get("appZones")
		)

		return _safe_http_json_response(response)

		except Exception as e:
		logger.exception("Failed to deploy service function")
		return {
		"error": "Deployment failed",
		"details": str(e)
		}, 500


		def delete_deployed_service_function(app_id: str):
		try:
		response = edgecloud_adapter.undeploy_app(app_id)
		return _safe_http_json_response(response)
		except Exception as e:
		logger.exception("Failed to undeploy service function")
		return {"error": "Undeploy failed", "details": str(e)}, 500


		def get_deployed_service_function(app_id: str):
		try:
		response = edgecloud_adapter.get_deployed_app(app_id=app_id)
		return _safe_http_json_response(response)
		except Exception as e:
		logger.exception("Failed to get deployed service function")
		return {"error": "Fetch failed", "details": str(e)}, 500


		def get_deployed_service_functions():
		try:
		response = edgecloud_adapter.get_all_deployed_apps()
		return _safe_http_json_response(response)
		except Exception as e:
		logger.exception("Failed to list deployed service functions")
		return {"error": "Fetch failed", "details": str(e)}, 500

		# ---------------------------------------------------------------------
		# Edge cloud nodes / zones
		# ---------------------------------------------------------------------

		def get_nodes():
		"""
		Returns the edge cloud zones.
		Normalizes the HTTP response and ensures edgeCloudRegion is always present.
		"""
		try:
		response = edgecloud_adapter.get_edge_cloud_zones()
		data, status = _safe_http_json_response(response)
		return _ensure_edge_cloud_region(data), status
		except Exception as e:
		logger.exception("Failed to get edge cloud zones")
		return {"error": "Fetch failed", "details": str(e)}, 500


		def node_details(node_id: str):
		"""
		Returns details for a specific edge cloud zone.
		Normalizes the HTTP response and ensures edgeCloudRegion is always present.
		"""
		try:
		response = edgecloud_adapter.get_edge_cloud_zones_details(zone_id=node_id)
		data, status = _safe_http_json_response(response)
		return _ensure_edge_cloud_region(data), status
		except Exception as e:
		logger.exception("Failed to get node details")
		return {"error": "Fetch failed", "details": str(e)}, 500

service-resource-manager-implementation/src/controllers/network_functions_controller.py

deleted100644 → 0

+0 −187

Changes for service-resource-manager-implementation/src/controllers/network_functions_controller.py: 0 added lines, 187 removed lines.
Original line number	Diff line number	Diff line
		from os import environ
		import logging
		import connexion
		from sunrise6g_opensdk.common.sdk import Sdk as sdkclient
		from sunrise6g_opensdk.network.core.schemas import RetrievalLocationRequest

		logger = logging.getLogger(__name__)
		logging.basicConfig(level=logging.DEBUG)
		network_adapter = None

		if environ.get('NETWORK_ADAPTER_NAME') and environ.get('NETWORK_ADAPTER_BASE_URL'):
		network_adapter_name = environ.get('NETWORK_ADAPTER_NAME')
		adapter_base_url = environ.get('NETWORK_ADAPTER_BASE_URL')
		scs_as_id = environ.get('SCS_AS_ID')
		network_adapter_specs = {
		'client_name': network_adapter_name,
		'base_url': adapter_base_url,
		'scs_as_id': scs_as_id,
		}
		print('Creating network adapter with specs: ', network_adapter_specs)
		adapters = sdkclient.create_adapters_from(adapter_specs={'network': network_adapter_specs})
		network_adapter = adapters.get("network")

		# else:
		# logging.error('Network adapter has not been specified! Aborting...')
		# sys.exit()


		def _safe_http_json_response(response):
		"""
		Normalize adapter responses:
		- list / dict → return directly (200)
		- Pydantic model → serialize via model_dump (200)
		- requests.Response → parse JSON body and status code safely
		"""
		if isinstance(response, (list, dict)):
		return response, 200

		if response is None:
		return {"error": "Adapter returned no response"}, 502

		if hasattr(response, 'model_dump'):
		return response.model_dump(mode='json', exclude_none=True), 200

		try:
		return response.json(), response.status_code
		except Exception as e:
		logger.exception("Failed to parse adapter HTTP response")
		return {"error": "Invalid response from adapter", "details": str(e)}, 502


		def create_qod_session():
		if connexion.request.is_json:
		try:
		if network_adapter is not None:
		response = network_adapter.create_qod_session(connexion.request.get_json())
		return _safe_http_json_response(response)
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500
		else:
		return {'error': 'Could not read JSON payload.'}, 400


		def get_qod_session(session_id: str):
		try:
		if network_adapter is not None:
		response = network_adapter.get_qod_session(session_id)
		return _safe_http_json_response(response)
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500


		def delete_qod_session(session_id: str):
		try:
		if network_adapter is not None:
		network_adapter.delete_qod_session(session_id)
		return "", 204
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500


		def create_traffic_influence_resource(body):
		try:
		if network_adapter is not None:
		response = network_adapter.create_traffic_influence_resource(traffic_influence_info=body)
		return _safe_http_json_response(response)
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500


		def delete_traffic_influence_resource(traffic_influence_id: str):
		try:
		if network_adapter is not None:
		response = network_adapter.delete_traffic_influence_resource(resource_id=traffic_influence_id)
		return _safe_http_json_response(response)
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500


		def get_traffic_influence_resource(traffic_influence_id: str):
		try:
		if network_adapter is not None:
		response = network_adapter.get_individual_traffic_influence_resource(resource_id=traffic_influence_id)
		return _safe_http_json_response(response)
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500


		def get_all_traffic_influence_resources():
		try:
		if network_adapter is not None:
		response = network_adapter.get_all_traffic_influence_resources()
		return _safe_http_json_response(response)
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500


		def retrieve_location():
		"""Retrieve the location of a device via the CAMARA Location Retrieval API."""

		if connexion.request.is_json:
		try:
		if network_adapter is not None:
		body = connexion.request.get_json()
		retrieve_location_request = RetrievalLocationRequest.model_validate(body)
		response = network_adapter.create_monitoring_event_subscription(retrieve_location_request)
		return _safe_http_json_response(response)
		else:
		return {
		"status": 503,
		"code": "UNAVAILABLE",
		"message": "Service unavailable: Network Adapter",
		}, 503
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500
		else:
		return {"error": "Could not read JSON payload."}, 400

SRM Refactor: FastAPI with new Domain Model and PostgreSQL adapter

Merge request reports