"""Service instance module.""" # Copyright 2022 Orange, Deutsche Telekom AG # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Iterator, Type, Union, Iterable, Optional from onapsdk.exceptions import StatusError, ParameterError from onapsdk.sdc.service import Service from onapsdk.so.deletion import ServiceDeletionRequest from onapsdk.so.instantiation import NetworkInstantiation, VnfInstantiation from onapsdk.utils.jinja import jinja_env from .instance import Instance from .network import NetworkInstance from .pnf import PnfInstance from .vnf import VnfInstance class ServiceInstance(Instance): # pylint: disable=too-many-instance-attributes """Service instanve class.""" def __init__(self, # pylint: disable=too-many-arguments, too-many-locals service_subscription: "ServiceSubscription", instance_id: str, instance_name: str = None, service_type: str = None, service_role: str = None, environment_context: str = None, workload_context: str = None, created_at: str = None, updated_at: str = None, resource_version: str = None, description: str = None, model_invariant_id: str = None, model_version_id: str = None, persona_model_version: str = None, widget_model_id: str = None, widget_model_version: str = None, bandwith_total: str = None, vhn_portal_url: str = None, service_instance_location_id: str = None, selflink: str = None, orchestration_status: str = None, input_parameters: str = None) -> None: """Service instance object initialization. Args: service_subscription (ServiceSubscription): service subscription which is belongs to instance_id (str): Uniquely identifies this instance of a service instance_name (str, optional): This field will store a name assigned to the service-instance. Defaults to None. service_type (str, optional): String capturing type of service. Defaults to None. service_role (str, optional): String capturing the service role. Defaults to None. environment_context (str, optional): This field will store the environment context assigned to the service-instance. Defaults to None. workload_context (str, optional): This field will store the workload context assigned to the service-instance. Defaults to None. created_at (str, optional): Create time of Network Service. Defaults to None. updated_at (str, optional): Last update of Network Service. Defaults to None. description (str, optional): Short description for service-instance. Defaults to None. model_invariant_id (str, optional): The ASDC model id for this resource or service model. Defaults to None. model_version_id (str, optional): The ASDC model version for this resource or service model. Defaults to None. persona_model_version (str, optional): The ASDC model version for this resource or service model. Defaults to None. widget_model_id (str, optional): The ASDC data dictionary widget model. This maps directly to the A&AI widget. Defaults to None. widget_model_version (str, optional): The ASDC data dictionary version of the widget model. This maps directly to the A&AI version of the widget. Defaults to None. bandwith_total (str, optional): Indicates the total bandwidth to be used for this service. Defaults to None. vhn_portal_url (str, optional): URL customers will use to access the vHN Portal. Defaults to None. service_instance_location_id (str, optional): An identifier that customers assign to the location where this service is being used. Defaults to None. resource_version (str, optional): Used for optimistic concurrency. Must be empty on create, valid on update and delete. Defaults to None. selflink (str, optional): Path to the controller object. Defaults to None. orchestration_status (str, optional): Orchestration status of this service. Defaults to None. input_parameters (str, optional): String capturing request parameters from SO to pass to Closed Loop. Defaults to None. """ super().__init__(resource_version=resource_version, model_invariant_id=model_invariant_id, model_version_id=model_version_id) self.service_subscription: "ServiceSubscription" = service_subscription self.instance_id: str = instance_id self.instance_name: str = instance_name self.service_type: str = service_type self.service_role: str = service_role self.environment_context: str = environment_context self.workload_context: str = workload_context self.created_at: str = created_at self.updated_at: str = updated_at self.description: str = description self.bandwith_total: str = bandwith_total self.vhn_portal_url: str = vhn_portal_url self.service_instance_location_id: str = service_instance_location_id self.selflink: str = selflink self.orchestration_status: str = orchestration_status self.input_parameters: str = input_parameters self.persona_model_version: str = persona_model_version self.widget_model_id: str = widget_model_id self.widget_model_version: str = widget_model_version self._sdc_service: Optional[Service] = None def __repr__(self) -> str: """Service instance object representation. Returns: str: Human readable service instance representation """ return (f"ServiceInstance(instance_id={self.instance_id}, " f"instance_name={self.instance_name})") def _get_related_instance(self, related_instance_class: Union[Type[NetworkInstance], Type[VnfInstance]], relationship_related_to_type: str) -> Iterator[\ Union[NetworkInstance, VnfInstance]]: """Iterate through related service instances. This is method which for given `relationship_related_to_type` creates iterator it iterate through objects which are related with service. Args: related_instance_class (Union[Type[NetworkInstance], Type[VnfInstance]]): Class object to create required object instances relationship_related_to_type (str): Has to be "generic-vnf" or "l3-network" Raises: ParameterError: relationship_related_to_type does not satisfy the requirements Yields: Iterator[ Union[NetworkInstance, VnfInstance]]: [description] """ if not relationship_related_to_type in ["l3-network", "generic-vnf", "pnf"]: msg = ( f'Invalid "relationship_related_to_type" value. ' f'Provided "{relationship_related_to_type}". ' f'Has to be "l3-network" or "generic-vnf".' ) raise ParameterError(msg) for relationship in self.relationships: if relationship.related_to == relationship_related_to_type: yield related_instance_class.create_from_api_response(\ self.send_message_json("GET", (f"Get {self.instance_id} " f"{related_instance_class.__class__.__name__}"), f"{self.base_url}{relationship.related_link}"), self) @classmethod def create(cls, service_subscription: "ServiceSubscription", # pylint: disable=too-many-arguments, too-many-locals instance_id: str, instance_name: str = None, service_type: str = None, service_role: str = None, environment_context: str = None, workload_context: str = None, created_at: str = None, updated_at: str = None, resource_version: str = None, description: str = None, model_invariant_id: str = None, model_version_id: str = None, persona_model_version: str = None, widget_model_id: str = None, widget_model_version: str = None, bandwith_total: str = None, vhn_portal_url: str = None, service_instance_location_id: str = None, selflink: str = None, orchestration_status: str = None, input_parameters: str = None): """Service instance creation. Args: service_subscription (ServiceSubscription): service subscription which is belongs to instance_id (str): Uniquely identifies this instance of a service instance_name (str, optional): This field will store a name assigned to the service-instance. Defaults to None. service_type (str, optional): String capturing type of service. Defaults to None. service_role (str, optional): String capturing the service role. Defaults to None. environment_context (str, optional): This field will store the environment context assigned to the service-instance. Defaults to None. workload_context (str, optional): This field will store the workload context assigned to the service-instance. Defaults to None. created_at (str, optional): Create time of Network Service. Defaults to None. updated_at (str, optional): Last update of Network Service. Defaults to None. description (str, optional): Short description for service-instance. Defaults to None. model_invariant_id (str, optional): The ASDC model id for this resource or service model. Defaults to None. model_version_id (str, optional): The ASDC model version for this resource or service model. Defaults to None. persona_model_version (str, optional): The ASDC model version for this resource or service model. Defaults to None. widget_model_id (str, optional): The ASDC data dictionary widget model. This maps directly to the A&AI widget. Defaults to None. widget_model_version (str, optional): The ASDC data dictionary version of the widget model. This maps directly to the A&AI version of the widget. Defaults to None. bandwith_total (str, optional): Indicates the total bandwidth to be used for this service. Defaults to None. vhn_portal_url (str, optional): URL customers will use to access the vHN Portal. Defaults to None. service_instance_location_id (str, optional): An identifier that customers assign to the location where this service is being used. Defaults to None. resource_version (str, optional): Used for optimistic concurrency. Must be empty on create, valid on update and delete. Defaults to None. selflink (str, optional): Path to the controller object. Defaults to None. orchestration_status (str, optional): Orchestration status of this service. Defaults to None. input_parameters (str, optional): String capturing request parameters from SO to pass to Closed Loop. Defaults to None. """ service_instance: "ServiceInstance" = cls( service_subscription, instance_id, instance_name, service_type, service_role, environment_context, workload_context, created_at, updated_at, resource_version, description, model_invariant_id, model_version_id, persona_model_version, widget_model_id, widget_model_version, bandwith_total, vhn_portal_url, service_instance_location_id, selflink, orchestration_status, input_parameters ) cls.send_message("PUT", f"Create service instance {instance_id} for "\ f"{service_subscription.service_type} service subscription", f"{service_subscription.url}/service-instances/service-instance/"\ f"{instance_id}", data=jinja_env() .get_template("aai_service_instance_create.json.j2") .render( service_instance=service_instance )) return service_instance @classmethod def get_all_url(cls, service_subscription: "ServiceSubscription") -> str: # pylint: disable=arguments-differ """Return an url to get all service instances for service subscription. Args: service_subscription (ServiceSubscription): Service subscription object Returns: str: Url to get all service instances for service subscription """ return f"{service_subscription.url}/service-instances/" @property def url(self) -> str: """Service instance resource URL. Returns: str: Service instance url """ return ( f"{self.service_subscription.url}/service-instances/service-instance/{self.instance_id}" ) @property def vnf_instances(self) -> Iterator[VnfInstance]: """Vnf instances associated with service instance. Returns iterator of VnfInstances representing VNF instantiated for that service Yields: VnfInstance: VnfInstance object """ return self._get_related_instance(VnfInstance, "generic-vnf") @property def network_instances(self) -> Iterator[NetworkInstance]: """Network instances associated with service instance. Returns iterator of NetworkInstance representing network instantiated for that service Yields: NetworkInstance: NetworkInstance object """ return self._get_related_instance(NetworkInstance, "l3-network") @property def pnfs(self) -> Iterator[PnfInstance]: """Pnfs associated with service instance. Returns iterator of PnfInstance representing pnfs instantiated for that service Yields: PnfInstance: PnfInstance object """ return self._get_related_instance(PnfInstance, "pnf") @property def sdc_service(self) -> Service: """Sdc service related with that instance. Sdc service model which was used to create that instance. Raises: ResourceNotFound: Service model not found """ if not self._sdc_service: self._sdc_service = Service.get_by_unique_uuid(self.model_invariant_id) return self._sdc_service @property def active(self) -> bool: """Information if service instance's orchestration status is active.""" return self.orchestration_status == "Active" def add_vnf(self, # pylint: disable=too-many-arguments vnf: "Vnf", line_of_business: "LineOfBusiness", platform: "Platform", cloud_region: "CloudRegion" = None, tenant: "Tenant" = None, vnf_instance_name: str = None, vnf_parameters: Iterable["InstantiationParameter"] = None, so_vnf: "SoServiceVnf" = None, a_la_carte: bool = True ) -> "VnfInstantiation": """Add vnf into service instance. Instantiate VNF. Args: vnf (Vnf): Vnf from service configuration to instantiate line_of_business (LineOfBusiness): LineOfBusiness to use in instantiation request platform (Platform): Platform to use in instantiation request cloud_region (CloudRegion, optional): Cloud region to use in instantiation request. Defaults to None. THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE TO USE IT!. tenant (Tenant, optional): Tenant to use in instnatiation request. Defaults to None. THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE TO USE IT!. vnf_instance_name (str, optional): VNF instantion name. If no value is provided it's going to be "Python_ONAP_SDK_vnf_instance_{str(uuid4())}". Defaults to None. vnf_parameters (Iterable[InstantiationParameter], optional): InstantiationParameter to be passed as "userParams". Defaults to None. so_vnf: (SoServiceVnf, optional): object with vnf instance parameters. Defaults to None. a_la_carte (bool): instantiation type for vnf. Defaults to True. Raises: StatusError: Service orchestration status is not "Active". Returns: VnfInstantiation: VnfInstantiation request object """ if not self.active: raise StatusError('Service orchestration status must be "Active"') if a_la_carte: return VnfInstantiation.instantiate_ala_carte( self, vnf, line_of_business, platform, cloud_region=cloud_region, tenant=tenant, vnf_instance_name=vnf_instance_name, vnf_parameters=vnf_parameters, sdc_service=self.sdc_service ) return VnfInstantiation.instantiate_macro( self, vnf, line_of_business, platform, cloud_region=cloud_region, tenant=tenant, vnf_instance_name=vnf_instance_name, vnf_parameters=vnf_parameters, so_vnf=so_vnf, sdc_service=self.sdc_service ) def add_network(self, # pylint: disable=too-many-arguments network: "Network", line_of_business: "LineOfBusiness", platform: "Platform", cloud_region: "CloudRegion" = None, tenant: "Tenant" = None, network_instance_name: str = None, subnets: Iterator["Subnet"] = None) -> "NetworkInstantiation": """Add network into service instance. Instantiate vl. Args: network (Network): Network from service configuration to instantiate line_of_business (LineOfBusiness): LineOfBusiness to use in instantiation request platform (Platform): Platform to use in instantiation request cloud_region (CloudRegion, optional): Cloud region to use in instantiation request. Defaults to None. THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE TO USE IT!. tenant (Tenant, optional): Tenant to use in instnatiation request. Defaults to None. THAT PROPERTY WILL BE REQUIRED IN ONE OF THE FUTURE RELEASE. REFACTOR YOUR CODE TO USE IT!. network_instance_name (str, optional): Network instantion name. If no value is provided it's going to be "Python_ONAP_SDK_network_instance_{str(uuid4())}". Defaults to None. Raises: StatusError: Service orchestration status is not "Active" Returns: NetworkInstantiation: NetworkInstantiation request object """ if not self.active: msg = f'Service orchestration status must be "Active"' raise StatusError(msg) return NetworkInstantiation.instantiate_ala_carte( self, network, line_of_business, platform, cloud_region=cloud_region, tenant=tenant, network_instance_name=network_instance_name, subnets=subnets ) def delete(self, a_la_carte: bool = True) -> "ServiceDeletionRequest": """Create service deletion request. Send a request to delete service instance Args: a_la_carte (boolean): deletion mode Returns: ServiceDeletionRequest: Deletion request object """ self._logger.debug("Delete %s service instance", self.instance_id) return ServiceDeletionRequest.send_request(self, a_la_carte)