Rename sdk invocation from universal catalog to sdk catalog (25f53340) · Commits · OOP / code / transformation-function-sdk

examples/basic_usage.py

+13 −6

Original line number	Diff line number	Diff line
		from src.common.universal_client_catalog import UniversalClientCatalog
		from src.common.sdk_catalog_client import SdkCatalogClient

		# "src.common.universal_catalog_client" would be equivalent to "sunrise6g_opensdk"


		def main():
		# Hardcoded client configuration; here we would expect to load the config
		# The module importing the SDK, loads the config
		client_specs = {
		"edgecloud": {
		"client_name": "i2edge",
		@@ -11,20 +13,25 @@ def main():
		# ,
		# "network": {
		# "client_name": "open5gs",
		# "base_url": "http://IP:PORT"
		# "base_url": "http://IP:PORT",
		# "scs_as_id": "id_example"
		# }
		}

		clients = UniversalClientCatalog.create_clients(client_specs)
		clients = SdkCatalogClient.create_clients(client_specs)

		edgecloud_client = clients.get("edgecloud")
		# network_client = clients.get("network")

		# Example usage
		print("Running test endpoint: get_edge_cloud_zones:")
		# Example of edgecloud client in action
		print("Testing edgecloud client function: get_edge_cloud_zones:")
		zones = edgecloud_client.get_edge_cloud_zones()
		print(zones)

		# # Example of network client in action
		# print("Testing network client function: EXAMPLE_FUNCTION:")
		# network_client.get_qod_session(session_id="example_session_id")


		if __name__ == "__main__":
		main()

src/common/universal_catalog.py→src/common/sdk_catalog.py

+21 −10

Original line number	Diff line number	Diff line
		@@ -11,6 +11,7 @@ def _edgecloud_catalog(client_name: str, base_url: str):
		edge_cloud_factory = {
		"aeros": lambda url: AerosClient(base_url=url),
		"i2edge": lambda url: I2EdgeClient(base_url=url),
		# TODO: uncomment when missing PiEdge's imports are added
		# "piedge": lambda url: PiEdgeClient(base_url=url),
		}
		try:
		@@ -21,35 +22,45 @@ def _edgecloud_catalog(client_name: str, base_url: str):
		)


		def _network_catalog(client_name: str, base_url: str):
		def _network_catalog(client_name: str, base_url: str, scs_as_id: str):
		network_factory = {
		"open5gs": lambda url: Open5GSClient(base_url=url),
		"oai": lambda url: OaiCoreClient(base_url=url),
		"open5gcore": lambda url: Open5GCoreClient(base_url=url),
		"open5gs": lambda url, scs_id: Open5GSClient(base_url=url, scs_as_id=scs_id),
		"oai": lambda url, scs_id: OaiCoreClient(base_url=url, scs_as_id=scs_id),
		"open5gcore": lambda url, scs_id: Open5GCoreClient(
		base_url=url, scs_as_id=scs_id
		),
		}
		try:
		return network_factory[client_name](base_url)
		return network_factory[client_name](base_url, scs_as_id)
		except KeyError:
		raise ValueError(
		f"Invalid network client '{client_name}'. Available: {list(network_factory)}"
		)


		class UniversalClientCatalog:
		# def _oran_catalog(client_name: str, base_url: str):
		# # TODO


		class SdkClientCatalog:
		_domain_factories = {
		"edgecloud": _edgecloud_catalog,
		"network": _network_catalog,
		# "oran": _oran_catalog,
		}

		@classmethod
		def get_client(cls, domain: str, client_name: str, base_url: str):
		def get_client(cls, domain: str, client_name: str, base_url: str, **kwargs):
		try:
		catalog = cls._domain_factories[domain]
		except KeyError:
		raise ValueError(
		f"Unsupported domain '{domain}'. Supported: {list(cls._domain_factories)}"
		)
		return catalog(client_name, base_url)


		universal_client_catalog = UniversalClientCatalog()
		if domain == "network":
		if "scs_as_id" not in kwargs:
		raise ValueError("Missing required 'scs_as_id' for network clients.")
		return catalog(client_name, base_url, kwargs["scs_as_id"])
		else:
		return catalog(client_name, base_url)

src/common/sdk_catalog_client.py

0 → 100644

+64 −0

Original line number	Diff line number	Diff line
		from typing import Dict

		from src.common.sdk_catalog import SdkClientCatalog


		class SdkCatalogClient:
		@staticmethod
		def create_clients(client_specs: Dict[str, Dict[str, str]]) -> Dict[str, object]:
		"""
		Create and return a dictionary of instantiated edgecloud/network/o-ran clients
		based on the provided specifications.

		Args:
		client_specs (dict): A dictionary where each key is the client's domain (e.g., 'edgecloud', 'network'),
		and each value is a dictionary containing:
		- 'client_name' (str): The specific name of the client (e.g., 'i2edge', 'open5gs').
		- 'base_url' (str): The base URL for the client's API.
		Additional parameters like 'scs_as_id' may also be included.

		Returns:
		dict: A dictionary where keys are the 'client_name' (str) and values are
		the instantiated client objects.

		Example:
		>>> from src.common.universal_client_catalog import UniversalCatalogClient
		>>>
		>>> client_specs_example = {
		>>> 'edgecloud': {
		>>> 'client_name': 'i2edge',
		>>> 'base_url': 'http://ip_edge_cloud:port',
		>>> 'additionalEdgeCloudParamater1': 'example'
		>>> },
		>>> 'network': {
		>>> 'client_name': 'open5gs',
		>>> 'base_url': 'http://ip_network:port',
		>>> 'additionalNetworkParamater1': 'example'
		>>> }
		>>> }
		>>>
		>>> clients = UniversalCatalogClient.create_clients(client_specs_example)
		>>> edgecloud_client = clients.get("edgecloud")
		>>> network_client = clients.get("network")
		>>>
		>>> edgecloud_client.get_edge_cloud_zones()
		>>> network_client.get_qod_session(session_id="example_session_id")
		"""
		universal_client_catalog = SdkClientCatalog()
		clients = {}

		for domain, config in client_specs.items():
		client_name = config["client_name"]
		base_url = config["base_url"]

		# Support of additional paramaters for specific clients
		kwargs = {
		k: v for k, v in config.items() if k not in ("client_name", "base_url")
		}

		client = universal_client_catalog.get_client(
		domain, client_name, base_url, **kwargs
		)
		clients[domain] = client

		return clients

src/common/universal_client_catalog.py

deleted100644 → 0

+0 −55

Original line number	Diff line number	Diff line
		from typing import Dict

		from src.common.universal_catalog import universal_client_catalog


		class UniversalClientCatalog:
		@staticmethod
		def create_clients(specs: Dict[str, Dict[str, str]]) -> Dict[str, object]:
		"""
		Create and return a dictionary of instantiated edgecloud/network/o-ran clients
		based on the provided specifications.

		Args:
		client_specs (list[dict]): A list of dictionaries, where each dictionary
		specifies a client to be created.
		Each client dictionary must contain:
		- 'domain' (str): The client's domain (e.g., 'edgecloud', 'network').
		- 'client_name' (str): The specific name of the client
		(e.g., 'i2edge', 'open5gs').
		- 'base_url' (str): The base URL for the client's API.
		Optional parameters (like 'scs_as_id') can also be included
		if required by the specific client's constructor.

		Returns:
		dict: A dictionary where keys are the 'client_name' (str) and values are
		the instantiated client objects.

		Example:
		>>> from src.common.universal_client_catalog import UniversalClientCatalog
		>>>
		>>> client_specs_example = [
		>>> {
		>>> 'domain': 'edgecloud',
		>>> 'client_name': 'i2edge',
		>>> 'base_url': 'http://localhost:8081',
		>>> 'description': 'i2edge client example.'
		>>> },
		>>> {
		>>> 'domain': 'network',
		>>> 'client_name': 'open5gs',
		>>> 'base_url': 'http://localhost:8084',
		>>> 'scs_as_id': 'my_unique_scs_id_example' # Example of optional parameter
		>>> }
		>>> ]
		>>>
		>>> clients = invoke_clients(client_specs_example)
		>>> # Access a client: i2edge_client = clients['i2edge']
		"""
		clients = {}
		for domain, config in specs.items():
		client_name = config["client_name"]
		base_url = config["base_url"]
		client = universal_client_catalog.get_client(domain, client_name, base_url)
		clients[domain] = client
		return clients

Admin message