Add custom error class for Network SDK (0c3f6b81) · Commits · OOP / code / transformation-function-sdk

src/network/clients/errors.py

0 → 100644

+2 −0

Original line number	Diff line number	Diff line
		class NetworkPlatformError(Exception):
		pass

src/network/clients/open5gs/client.py

0 → 100644

+57 −0

Original line number	Diff line number	Diff line
		from typing import Dict
		from src import logger
		from src.network.core.network_interface import NetworkManagementInterface
		from . import common
		from . import schemas

		log = logger.get_logger(__name__)

		class Open5GSClient(NetworkManagementInterface):
		"""
		This client implements the NetworkManagementInterface and translates the
		CAMARA APIs into specific HTTP requests understandable by the Open5GS NEF API.

		Invloved partners and their roles in this implementation:
		- I2CAT: Responsible for the CAMARA QoD API and its mapping to the 3GPP AsSessionWithQoS API exposed by Open5GS NEF.
		- NCSRD: Responsible for the CAMARA Location API and its mapping to the 3GPP Monitoring Even API exposed Open5GS NEF.
		"""

		def __init__(self, base_url: str, scs_as_id: str):
		"""
		Initializes the Open5GS Client.
		"""
		try:
		self.base_url = base_url
		self.scs_as_id = scs_as_id
		log.info(f"Initialized Open5GSClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}")
		except Exception as e:
		log.error(f"Failed to initialize Open5GSClient: {e}")
		raise e


		# --- Implementation of NetworkManagementInterface methods ---
		def create_qod_session(self, session_info: Dict) -> Dict:
		"""
		Creates a QoD session based on the CAMARA QoD API input.
		Maps the CAMARA QoD POST /sessions to Open5GS NEF POST /{scsAsId}/subscriptions.
		"""
		pass


		def get_qod_session(self, session_id: str) -> Dict:
		"""
		Retrieves a specific Open5GS QoS Subscription details.
		Maps CAMARA QoD GET /sessions/{sessionId} to Open5GS NEF GET /{scsAsId}/subscriptions/{subscriptionId}.
		"""
		pass


		def delete_qod_session(self, session_id: str) -> None:
		"""
		Deletes a specific Open5GS QoS Subscription.
		Maps CAMARA QoD DELETE /sessions/{sessionId} to Open5GS NEF DELETE /{scsAsId}/subscriptions/{subscriptionId}.
		"""
		pass

src/network/clients/open5gs/common.py

0 → 100644

+38 −0

Original line number	Diff line number	Diff line
		# Common utilities (errors, HTTP helpers) used by the Open5GS client implementation (client.py).
		import json
		from typing import Optional

		import requests
		from pydantic import BaseModel, ValidationError
		from src.network.clients.errors import NetworkPlatformError
		from src import logger

		log = logger.get_logger(__name__)


		class Open5GSError(NetworkPlatformError):
		pass

		class Open5GSErrorResponse(BaseModel):
		message: str
		detail: dict

		# --- HTTP Request Helper Functions ---
		def open5gs_post(url: str, model_payload: BaseModel) -> dict:
		"""
		Placeholder for the POST request function."""
		pass

		def open5gs_get(url: str, params: Optional[dict] = None) -> dict:
		"""
		Placeholder for the GET request function.
		"""
		pass

		def open5gs_delete(url: str) -> None:
		"""
		Placeholder for the DELETE request function.
		"""
		pass

src/network/clients/open5gs/schemas.py

0 → 100644

+19 −0

Original line number	Diff line number	Diff line
		# This file defines the Pydantic models that represent the data structures (schemas)
		# for the requests sent to and responses received from the Open5GS NEF API,
		# specifically focusing on the APIs needed to support CAMARA QoD.

		from pydantic import BaseModel

		# Dummy examples of Pydantic models for the Open5GS NEF API.
		class Open5GSQoSSubscription(BaseModel):
		"""
		Represents the payload for creating a QoS subscription in Open5GS.
		"""
		pass

		class CamaraQoDSessionInfo(BaseModel):
		"""
		Represents the input data for creating a QoD session.
		"""
		pass

src/network/core/network_factory.py

0 → 100644

+59 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# -- coding: utf-8 --
		##
		# Copyright 2025-present by Software Networks Area, i2CAT.
		# All rights reserved.
		#
		# This file is part of the Open SDK
		#
		# Contributors:
		# - Reza Mosahebfard (reza.mosahebfard@i2cat.net)
		##
		from __future__ import annotations

		from typing import TYPE_CHECKING

		from src.network.clients.open5gs.client import Open5GSClient
		from src.network.clients.oai.client import OaiNefClient
		from src.network.clients.open5gcore.client import Open5GCoreClient

		if TYPE_CHECKING:
		from .network_interface import NetworkManagementInterface

		class NetworkClientFactory:
		"""
		Factory class for creating Network Management Clients.
		"""

		@staticmethod
		def create_network_client(client_name: str, base_url: str) -> NetworkManagementInterface:
		"""
		Creates and returns an instance of the specified Network Client.
		"""
		try:
		constructor = NetworkClientFactory.network_client_constructors[client_name]
		network_client_instance = constructor(base_url)
		return network_client_instance
		except KeyError:
		# Get the list of supported client names
		supported_clients = list(NetworkClientFactory.network_client_constructors.keys())
		raise ValueError(
		f"Invalid network client name: '{client_name}'. "
		f"Supported clients are: {', '.join(supported_clients)}"
		)

		class NetworkClientTypes:
		"""
		Class for creating Network Clients.
		"""

		OPEN5GS = "open5gs"
		OAI = "oai"
		OPEN5GCORE = "open5gcore"

		# --- Dictionary mapping type constants to constructors ---
		network_types = {
		OPEN5GS: lambda url: Open5GSClient(base_url=url),
		OAI: lambda url: OaiNefClient(base_url=url),
		OPEN5GCORE: lambda url: Open5GCoreClient(base_url=url),
		}

Admin message