WIP - adding endpoints and datamodels of endpoints of H-RAT (c32fa219) · Commits · TFS / controller

src/pathcompextended/connector/hrat.py

+513 −36

Original line number	Diff line number	Diff line
		@@ -14,14 +14,200 @@

		import json
		import logging
		from typing import Dict, List
		from typing import Dict, List, Optional, Any
		from pydantic import BaseModel, Field

		from pathcompextended.connector.api_client import APIClient
		import pathcompextended.Config as Config

		# Import base models
		from pathcompextended.model.network_context import NetworkContextRequest, Topology
		from pathcompextended.model.transport_optical_slice import (
		TransportOpticalSliceRequest,
		TransportOpticalSliceResponse,
		)
		from pathcompextended.model.transport_network_slice_l3 import (
		TransportNetworkSliceL3Request,
		TransportNetworkSliceL3Response,
		)
		from pathcompextended.model.health import HealthCheckResponse

		LOGGER = logging.getLogger(__name__)


		# ============================================================================
		# NETWORK CONTEXT MODELS
		# ============================================================================

		class NetworkContextPostRequest(NetworkContextRequest):
		"""Request model for POST /network-context."""
		pass


		class NetworkContextPostResponse(BaseModel):
		"""Response model for POST /network-context."""
		message: str
		total_topologies: int = Field(..., alias="total-topologies")

		class Config:
		populate_by_name = True


		class NetworkContextGetResponse(BaseModel):
		"""Response model for GET /network-context."""
		data: List[Topology]
		total: Optional[int] = None
		message: Optional[str] = None


		class NetworkContextGetByControllerResponse(BaseModel):
		"""Response model for GET /network-context/{controllerId}."""
		data: List[Topology]
		total: Optional[int] = None
		message: Optional[str] = None


		class NetworkContextDeleteResponse(BaseModel):
		"""Response model for DELETE /network-context/{controllerId}."""
		deleted_count: int = Field(..., alias="deleted-count")
		message: str

		class Config:
		populate_by_name = True


		# ============================================================================
		# TRANSPORT OPTICAL SLICE MODELS
		# ============================================================================

		class TransportOpticalSlicePostRequest(TransportOpticalSliceRequest):
		"""Request model for POST /transport-optical-slice."""
		pass


		class TransportOpticalSlicePostResponse(BaseModel):
		"""Response model for POST /transport-optical-slice (returns UUID string)."""
		# The API returns a UUID string directly, but we'll parse it
		optical_slice_uuid: str


		class TransportOpticalSliceGetResponse(BaseModel):
		"""Response model for GET /transport-optical-slice/{uuid}."""
		data: TransportOpticalSliceResponse
		message: str


		class TransportOpticalSliceGetAllResponse(BaseModel):
		"""Response model for GET /transport-optical-slice."""
		total: int
		data: List[TransportOpticalSliceResponse]
		message: str


		class TransportOpticalSliceDeleteResponse(BaseModel):
		"""Response model for DELETE /transport-optical-slice/{uuid}."""
		deleted_optical_slice_uuid: str = Field(..., alias="deleted-optical-slice-uuid")
		deleted_transport_network_slice_l3_uuids: List[str] = Field(
		..., alias="deleted-transport-network-slice-l3-uuids"
		)
		message: str

		class Config:
		populate_by_name = True


		class TransportOpticalSliceDeleteAllResponse(BaseModel):
		"""Response model for DELETE /transport-optical-slice."""
		deleted_optical_slice_uuids: List[str] = Field(..., alias="deleted-optical-slice-uuids")
		deleted_transport_network_slice_l3_uuids: List[str] = Field(
		..., alias="deleted-transport-network-slice-l3-uuids"
		)
		message: str

		class Config:
		populate_by_name = True


		# ============================================================================
		# TRANSPORT NETWORK SLICE L3 MODELS
		# ============================================================================

		class TransportNetworkSliceL3PostRequest(TransportNetworkSliceL3Request):
		"""Request model for POST /transport-network-slice-l3."""
		pass


		class TransportNetworkSliceL3PostResponse(BaseModel):
		"""Response model for POST /transport-network-slice-l3 (returns UUID string)."""
		# The API returns a UUID string directly, but we'll parse it
		transport_network_slice_l3_uuid: str


		class TransportNetworkSliceL3GetResponse(BaseModel):
		"""Response model for GET /transport-network-slice-l3/{uuid}."""
		data: TransportNetworkSliceL3Response
		message: str


		class TransportNetworkSliceL3GetAllResponse(BaseModel):
		"""Response model for GET /transport-network-slice-l3."""
		total: int
		data: List[TransportNetworkSliceL3Response]
		message: str


		class TransportNetworkSliceL3DeleteResponse(BaseModel):
		"""Response model for DELETE /transport-network-slice-l3/{uuid}."""
		deleted_count: int = Field(..., alias="deleted-count")
		deleted_transport_network_slice_l3_uuids: List[str] = Field(
		..., alias="deleted-transport-network-slice-l3-uuids"
		)
		message: str

		class Config:
		populate_by_name = True


		class TransportNetworkSliceL3DeleteAllResponse(BaseModel):
		"""Response model for DELETE /transport-network-slice-l3."""
		deleted_count: int = Field(..., alias="deleted-count")
		deleted_transport_network_slice_l3_uuids: List[str] = Field(
		..., alias="deleted-transport-network-slice-l3-uuids"
		)
		message: str

		class Config:
		populate_by_name = True


		# ============================================================================
		# HEALTH CHECK MODELS
		# ============================================================================

		class HealthCheckGetResponse(HealthCheckResponse):
		"""Response model for GET /health."""
		pass


		# ============================================================================
		# EXTERNAL KPIs MODELS
		# ============================================================================

		class ExternalKpisPostRequest(BaseModel):
		"""Request model for POST /external-kpis."""
		# Accepts arbitrary JSON structure
		data: Dict[str, Any]


		class ExternalKpisPostResponse(BaseModel):
		"""Response model for POST /external-kpis."""
		message: str


		# ============================================================================
		# HRAT CLIENT CLASS
		# ============================================================================

		class HRAT:
		"""
		Client for interacting with H-RAT (Hybrid Resource Allocation Tool) external service.
		@@ -46,72 +232,334 @@ class HRAT:

		LOGGER.debug("HRAT initialized with url: {:s}".format(url))

		def network_context(self, networks: List[Dict]):
		# ========================================================================
		# NETWORK CONTEXT METHODS
		# ========================================================================

		def network_context(self, request: NetworkContextPostRequest) -> NetworkContextPostResponse:
		"""
		Send network context to H-RAT.

		Args:
		networks: List of network topology dictionaries
		request: NetworkContextPostRequest with topologies

		Returns:
		Response object from H-RAT API
		NetworkContextPostResponse with message and total-topologies
		"""
		LOGGER.debug("Sending network context to HRAT")

		endpoint = f"{self.base_api_url}/network-context"

		body = {
		"topologies": [
		{
		"topology-id": topology.get('topology_id', ''),
		"topology-type": topology.get('topology_type', ''),
		"controller-id": topology.get('controller_id', ''),
		"raw-json-topology": json.dumps(topology.get('json_topology', {})),
		} for topology in networks
		]
		}
		# Convert Pydantic model to dict for API call
		body = request.model_dump(by_alias=True, exclude_none=True)

		res = self.api.post(endpoint, body)

		if res.status_code != 200:
		LOGGER.error(f"Failed to create network context: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return NetworkContextPostResponse(**res.json())

		def get_network_context(self) -> NetworkContextGetResponse:
		"""
		Get all network contexts from H-RAT.

		Returns:
		NetworkContextGetResponse with list of topologies
		"""
		LOGGER.debug("Getting all network contexts from HRAT")

		endpoint = f"{self.base_api_url}/network-context"
		res = self.api.get(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to get network contexts: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return res
		return NetworkContextGetResponse(**res.json())

		def transport_optical_slice(self, network_slice: Dict):
		def get_network_context_by_controller(self, controller_id: str) -> NetworkContextGetByControllerResponse:
		"""
		Get network contexts by controller ID.

		Args:
		controller_id: Controller identifier

		Returns:
		NetworkContextGetByControllerResponse with list of topologies
		"""
		LOGGER.debug(f"Getting network contexts for controller {controller_id}")

		endpoint = f"{self.base_api_url}/network-context/{controller_id}"
		res = self.api.get(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to get network contexts: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return NetworkContextGetByControllerResponse(**res.json())

		def delete_network_context_by_controller(self, controller_id: str) -> NetworkContextDeleteResponse:
		"""
		Delete network contexts by controller ID.

		Args:
		controller_id: Controller identifier

		Returns:
		NetworkContextDeleteResponse with deletion result
		"""
		LOGGER.debug(f"Deleting network contexts for controller {controller_id}")

		endpoint = f"{self.base_api_url}/network-context/{controller_id}"
		res = self.api.delete(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to delete network contexts: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return NetworkContextDeleteResponse(**res.json())

		# ========================================================================
		# TRANSPORT OPTICAL SLICE METHODS
		# ========================================================================

		def transport_optical_slice(self, request: TransportOpticalSlicePostRequest) -> TransportOpticalSlicePostResponse:
		"""
		Request transport optical slice computation from H-RAT.

		Args:
		network_slice: Network slice dictionary with IETF network slice definition
		request: TransportOpticalSlicePostRequest with IETF network slice definition

		Returns:
		TransportOpticalSlicePostResponse with optical-slice-uuid
		"""
		LOGGER.debug("Requesting transport optical slice to HRAT")

		endpoint = f"{self.base_api_url}/transport-optical-slice"

		# Convert Pydantic model to dict for API call
		body = request.model_dump(by_alias=True, exclude_none=True)

		res = self.api.post(endpoint, body)

		if res.status_code != 200:
		LOGGER.error(f"Failed to create transport optical slice: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")

		# The API returns a UUID string directly
		uuid_response = res.text.strip('"')
		return TransportOpticalSlicePostResponse(optical_slice_uuid=uuid_response)

		def get_transport_optical_slice(self, uuid: str) -> TransportOpticalSliceGetResponse:
		"""
		Get transport optical slice by UUID.

		Args:
		uuid: Optical slice UUID

		Returns:
		TransportOpticalSliceGetResponse with slice data
		"""
		LOGGER.debug(f"Getting transport optical slice {uuid}")

		endpoint = f"{self.base_api_url}/transport-optical-slice/{uuid}"
		res = self.api.get(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to get transport optical slice: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return TransportOpticalSliceGetResponse(**res.json())

		def get_all_transport_optical_slices(self) -> TransportOpticalSliceGetAllResponse:
		"""
		Get all transport optical slices.

		Returns:
		TransportOpticalSliceGetAllResponse with list of slices
		"""
		LOGGER.debug("Getting all transport optical slices")

		endpoint = f"{self.base_api_url}/transport-optical-slice"
		res = self.api.get(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to get transport optical slices: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return TransportOpticalSliceGetAllResponse(**res.json())

		def delete_transport_optical_slice(self, uuid: str) -> TransportOpticalSliceDeleteResponse:
		"""
		Delete transport optical slice by UUID.

		Args:
		uuid: Optical slice UUID

		Returns:
		TransportOpticalSliceDeleteResponse with deletion result
		"""
		LOGGER.debug(f"Deleting transport optical slice {uuid}")

		endpoint = f"{self.base_api_url}/transport-optical-slice/{uuid}"
		res = self.api.delete(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to delete transport optical slice: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return TransportOpticalSliceDeleteResponse(**res.json())

		def delete_all_transport_optical_slices(self) -> TransportOpticalSliceDeleteAllResponse:
		"""
		Delete all transport optical slices.

		Returns:
		Response object from H-RAT API
		TransportOpticalSliceDeleteAllResponse with deletion results
		"""
		LOGGER.debug(f"Requesting transport optical slice to HRAT")
		LOGGER.debug("Deleting all transport optical slices")

		endpoint = f"{self.base_api_url}/transport-optical-slice"
		res = self.api.post(endpoint, network_slice)
		res = self.api.delete(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to delete all transport optical slices: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return res
		return TransportOpticalSliceDeleteAllResponse(**res.json())

		# ========================================================================
		# TRANSPORT NETWORK SLICE L3 METHODS
		# ========================================================================

		def transport_network_slice_l3(self, network_slice: Dict):
		def transport_network_slice_l3(self, request: TransportNetworkSliceL3PostRequest) -> TransportNetworkSliceL3PostResponse:
		"""
		Request transport network slice L3 computation from H-RAT.

		Args:
		network_slice: Network slice dictionary with IETF network slice definition
		request: TransportNetworkSliceL3PostRequest with IETF network slice definition

		Returns:
		Response object from H-RAT API
		TransportNetworkSliceL3PostResponse with transport-network-slice-l3-uuid
		"""
		LOGGER.debug(f"Requesting transport network slice L3 to HRAT")
		LOGGER.debug("Requesting transport network slice L3 to HRAT")

		endpoint = f"{self.base_api_url}/transport-network-slice-l3"
		res = self.api.post(endpoint, network_slice)

		# Convert Pydantic model to dict for API call
		body = request.model_dump(by_alias=True, exclude_none=True)

		res = self.api.post(endpoint, body)

		if res.status_code not in [200, 409]:
		LOGGER.error(f"Failed to create transport network slice L3: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return res

		def health(self):
		# The API returns a UUID string directly
		uuid_response = res.text.strip('"')
		return TransportNetworkSliceL3PostResponse(transport_network_slice_l3_uuid=uuid_response)

		def get_transport_network_slice_l3(self, uuid: str) -> TransportNetworkSliceL3GetResponse:
		"""
		Get transport network slice L3 by UUID.

		Args:
		uuid: Transport network slice L3 UUID

		Returns:
		TransportNetworkSliceL3GetResponse with slice data
		"""
		LOGGER.debug(f"Getting transport network slice L3 {uuid}")

		endpoint = f"{self.base_api_url}/transport-network-slice-l3/{uuid}"
		res = self.api.get(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to get transport network slice L3: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return TransportNetworkSliceL3GetResponse(**res.json())

		def get_all_transport_network_slices_l3(self) -> TransportNetworkSliceL3GetAllResponse:
		"""
		Get all transport network slices L3.

		Returns:
		TransportNetworkSliceL3GetAllResponse with list of slices
		"""
		LOGGER.debug("Getting all transport network slices L3")

		endpoint = f"{self.base_api_url}/transport-network-slice-l3"
		res = self.api.get(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to get transport network slices L3: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return TransportNetworkSliceL3GetAllResponse(**res.json())

		def delete_transport_network_slice_l3(self, uuid: str) -> TransportNetworkSliceL3DeleteResponse:
		"""
		Delete transport network slice L3 by UUID.

		Args:
		uuid: Transport network slice L3 UUID

		Returns:
		TransportNetworkSliceL3DeleteResponse with deletion result
		"""
		LOGGER.debug(f"Deleting transport network slice L3 {uuid}")

		endpoint = f"{self.base_api_url}/transport-network-slice-l3/{uuid}"
		res = self.api.delete(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to delete transport network slice L3: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return TransportNetworkSliceL3DeleteResponse(**res.json())

		def delete_all_transport_network_slices_l3(self) -> TransportNetworkSliceL3DeleteAllResponse:
		"""
		Delete all transport network slices L3.

		Returns:
		TransportNetworkSliceL3DeleteAllResponse with deletion results
		"""
		LOGGER.debug("Deleting all transport network slices L3")

		endpoint = f"{self.base_api_url}/transport-network-slice-l3"
		res = self.api.delete(endpoint)

		if res.status_code != 200:
		LOGGER.error(f"Failed to delete all transport network slices L3: {res.status_code} - {res.text}")
		res.raise_for_status()

		LOGGER.debug(f"Response from HRAT: {res.status_code} - {res.text}")
		return TransportNetworkSliceL3DeleteAllResponse(**res.json())

		# ========================================================================
		# HEALTH CHECK METHODS
		# ========================================================================

		def health(self) -> bool:
		"""
		Check H-RAT service health.

		@@ -120,10 +568,24 @@ class HRAT:
		"""
		LOGGER.debug("HRAT health check")
		res = self.api.get('/health')
		LOGGER.debug(f"HRAT health check result: {res.status_code}")
		return res.status_code == 200

		def external_kpis(self, kpis: Dict):
		if res.status_code == 200:
		try:
		health_response = HealthCheckGetResponse(**res.json())
		LOGGER.debug(f"HRAT health check result: {health_response.status} - {health_response.timestamp}")
		return health_response.status == "OK"
		except Exception as e:
		LOGGER.warning(f"Failed to parse health response: {e}")
		return True # If status is 200, consider it healthy
		else:
		LOGGER.warning(f"HRAT health check failed: {res.status_code}")
		return False

		# ========================================================================
		# EXTERNAL KPIs METHODS
		# ========================================================================

		def external_kpis(self, kpis: Dict[str, Any]) -> ExternalKpisPostResponse:
		"""
		Send external KPIs to H-RAT.

		@@ -131,12 +593,27 @@ class HRAT:
		kpis: Dictionary containing KPI metrics

		Returns:
		Response object from H-RAT API
		ExternalKpisPostResponse with confirmation message
		"""
		LOGGER.debug(f"Sending external KPIs to HRAT")
		LOGGER.debug("Sending external KPIs to HRAT")

		endpoint = f"{self.base_api_url}/external-kpis"
		res = self.api.post(endpoint, kpis)

		# Create request model
		request = ExternalKpisPostRequest(data=kpis)
		body = request.model_dump(by_alias=True, exclude_none=True)

		res = self.api.post(endpoint, body)

		if res.status_code != 200:
		LOGGER.warning(f"Failed to send external KPIs: {res.status_code} - {res.text}")
		# Don't raise, KPIs are not critical
		return ExternalKpisPostResponse(message=f"Failed: {res.text}")

		LOGGER.debug(f"Response from HRAT external-kpis: {res.status_code} - {res.text}")
		return res

		try:
		return ExternalKpisPostResponse(**res.json())
		except Exception as e:
		LOGGER.warning(f"Failed to parse KPIs response: {e}")
		return ExternalKpisPostResponse(message=res.text)

src/pathcompextended/model/NetworkContext.py

deleted100644 → 0

+0 −4

Original line number	Diff line number	Diff line
		from pydantic import BaseModel

		class NetworkContext(BaseModel):
		pass
		No newline at end of file

src/pathcompextended/model/TransportNetworkSliceL3.py

deleted100644 → 0

+0 −4

Original line number	Diff line number	Diff line
		from pydantic import BaseModel

		class NetworkContext(BaseModel):
		pass
		No newline at end of file

src/pathcompextended/model/TransportOpticalSlice.py

deleted100644 → 0

+0 −4

Original line number	Diff line number	Diff line
		from pydantic import BaseModel

		class NetworkContext(BaseModel):
		pass
		No newline at end of file

src/pathcompextended/model/init.py

+79 −0

File changed.

Preview size limit exceeded, changes collapsed.