Refactored adapters_factory.py for lazy adapter loading. (2222cd96) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/common/adapters_factory.py

+75 −62

Original line number	Diff line number	Diff line
		@@ -5,85 +5,97 @@
		# Contributors:
		# - Adrián Pino Martínez (adrian.pino@i2cat.net)
		##
		import importlib
		from typing import Dict, Tuple

		from sunrise6g_opensdk.edgecloud.adapters.aeros.client import (
		EdgeApplicationManager as AerosClient,
		)
		from sunrise6g_opensdk.edgecloud.adapters.i2edge.client import (
		EdgeApplicationManager as I2EdgeClient,
		)
		from sunrise6g_opensdk.edgecloud.adapters.kubernetes.client import (
		EdgeApplicationManager as kubernetesClient,
		)
		from sunrise6g_opensdk.network.adapters.oai.client import (
		NetworkManager as OaiCoreClient,
		)
		from sunrise6g_opensdk.network.adapters.open5gcore.client import (
		NetworkManager as Open5GCoreClient,
		)
		from sunrise6g_opensdk.network.adapters.open5gs.client import (
		NetworkManager as Open5GSClient,
		)
		from sunrise6g_opensdk.oran.adapters.i2cat_ric.client import (
		OranManager as OranManageri2CAT,
		)
		AdapterSpec = Tuple[str, str]

		_EDGE_CLOUD_CATALOG: Dict[str, AdapterSpec] = {
		"aeros": (
		"sunrise6g_opensdk.edgecloud.adapters.aeros.client",
		"EdgeApplicationManager",
		),
		"i2edge": (
		"sunrise6g_opensdk.edgecloud.adapters.i2edge.client",
		"EdgeApplicationManager",
		),
		"kubernetes": (
		"sunrise6g_opensdk.edgecloud.adapters.kubernetes.client",
		"EdgeApplicationManager",
		),
		}

		def _edgecloud_adapters_factory(client_name: str, base_url: str, **kwargs):
		if client_name == "i2edge":
		if "flavour_id" not in kwargs:
		raise ValueError("Missing required 'flavour_id' for i2edge client.")
		_NETWORK_CATALOG: Dict[str, AdapterSpec] = {
		"open5gs": (
		"sunrise6g_opensdk.network.adapters.open5gs.client",
		"NetworkManager",
		),
		"oai": (
		"sunrise6g_opensdk.network.adapters.oai.client",
		"NetworkManager",
		),
		"open5gcore": (
		"sunrise6g_opensdk.network.adapters.open5gcore.client",
		"NetworkManager",
		),
		}

		edge_cloud_factory = {
		"aeros": lambda url, kw: AerosClient(base_url=url, kw),
		"i2edge": lambda url, kw: I2EdgeClient(base_url=url, kw),
		"kubernetes": lambda url, kw: kubernetesClient(base_url=url, kw),
		_ORAN_CATALOG: Dict[str, AdapterSpec] = {
		"i2cat_ric": (
		"sunrise6g_opensdk.oran.adapters.i2cat_ric.client",
		"OranManager",
		),
		}


		def _resolve_adapter(module_path: str, attr_name: str):
		module = importlib.import_module(module_path)
		return getattr(module, attr_name)


		def _instantiate_from_catalog(
		catalog: Dict[str, AdapterSpec], client_name: str, base_url: str, **kwargs
		):
		try:
		return edge_cloud_factory[client_name](base_url, **kwargs)
		except KeyError:
		module_path, attr_name = catalog[client_name]
		except KeyError as exc:
		raise ValueError(
		f"Invalid edgecloud client '{client_name}'. Available: {list(edge_cloud_factory)}"
		)
		f"Invalid client '{client_name}'. Available: {list(catalog)}"
		) from exc

		client_cls = _resolve_adapter(module_path, attr_name)
		return client_cls(base_url=base_url, **kwargs)


		def _edgecloud_adapters_factory(client_name: str, base_url: str, **kwargs):
		if client_name == "i2edge" and "flavour_id" not in kwargs:
		raise ValueError("Missing required 'flavour_id' for i2edge client.")

		return _instantiate_from_catalog(_EDGE_CLOUD_CATALOG, client_name, base_url, **kwargs)


		def _network_adapters_factory(client_name: str, base_url: str, **kwargs):
		if "scs_as_id" not in kwargs:
		raise ValueError("Missing required 'scs_as_id' for network adapters.")
		scs_as_id = kwargs.pop("scs_as_id")

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


		def _oran_adapters_factory(client_name: str, base_url: str, **kwargs):
		if "scs_as_id" not in kwargs:
		raise ValueError("Missing required 'scs_as_id' for network adapters.")
		scs_as_id = kwargs.pop("scs_as_id")
		raise ValueError("Missing required 'scs_as_id' for ORAN adapters.")

		oran_factory = {
		"i2cat_ric": lambda url, scs_id, **kw: OranManageri2CAT(
		base_url=url, scs_as_id=scs_id, **kw
		),
		}
		try:
		return oran_factory[client_name](base_url, scs_as_id, **kwargs)
		except KeyError:
		raise ValueError(f"Invalid Oran client '{client_name}'. Available: {list(oran_factory)}")
		scs_as_id = kwargs.pop("scs_as_id")
		return _instantiate_from_catalog(
		_ORAN_CATALOG, client_name, base_url, scs_as_id=scs_as_id, **kwargs
		)


		class AdaptersFactory:

		_domain_factories = {
		"edgecloud": _edgecloud_adapters_factory,
		"network": _network_adapters_factory,
		@@ -96,8 +108,9 @@ class AdaptersFactory:
		):
		try:
		catalog = cls._domain_factories[domain]
		except KeyError:
		except KeyError as exc:
		raise ValueError(
		f"Unsupported domain '{domain}'. Supported: {list(cls._domain_factories)}"
		)
		) from exc

		return catalog(client_name, base_url, **kwargs)

Admin message