src/onapsdk/aai/business/customer.py
"""AAI business module."""
from dataclasses import dataclass
from typing import Iterable, Iterator, Optional
from urllib.parse import urlencode
from onapsdk.utils.jinja import jinja_env
from onapsdk.exceptions import APIError, ParameterError, ResourceNotFound
from ..aai_element import AaiResource, Relationship
from ..cloud_infrastructure.cloud_region import CloudRegion
from .service import ServiceInstance
@dataclass
class ServiceSubscriptionCloudRegionTenantData:
"""Dataclass to store cloud regions and tenants data for service subscription."""
cloud_owner: str = None
cloud_region_id: str = None
tenant_id: str = None
@dataclass
class ServiceSubscription(AaiResource):
"""Service subscription class."""
service_type: str
resource_version: str
customer: "Customer"
def __init__(self, customer: "Customer", service_type: str, resource_version: str) -> None:
"""Service subscription object initialization.
Args:
customer (Customer): Customer object
service_type (str): Service type
resource_version (str): Service subscription resource version
"""
super().__init__()
self.customer: "Customer" = customer
self.service_type: str = service_type
self.resource_version: str = resource_version
def _get_service_instance_by_filter_parameter(self,
filter_parameter_name: str,
filter_parameter_value: str) -> ServiceInstance:
"""Call a request to get service instance with given filter parameter and value.
Args:
filter_parameter_name (str): Name of parameter to filter
filter_parameter_value (str): Value of filter parameter
Returns:
ServiceInstance: ServiceInstance object
"""
service_instance: dict = self.send_message_json(
"GET",
f"Get service instance with {filter_parameter_value} {filter_parameter_name}",
f"{self.url}/service-instances?{filter_parameter_name}={filter_parameter_value}"
)["service-instance"][0]
return ServiceInstance(
service_subscription=self,
instance_id=service_instance.get("service-instance-id"),
instance_name=service_instance.get("service-instance-name"),
service_type=service_instance.get("service-type"),
service_role=service_instance.get("service-role"),
environment_context=service_instance.get("environment-context"),
workload_context=service_instance.get("workload-context"),
created_at=service_instance.get("created-at"),
updated_at=service_instance.get("updated-at"),
description=service_instance.get("description"),
model_invariant_id=service_instance.get("model-invariant-id"),
model_version_id=service_instance.get("model-version-id"),
persona_model_version=service_instance.get("persona-model-version"),
widget_model_id=service_instance.get("widget-model-id"),
widget_model_version=service_instance.get("widget-model-version"),
bandwith_total=service_instance.get("bandwidth-total"),
vhn_portal_url=service_instance.get("vhn-portal-url"),
service_instance_location_id=service_instance.get("service-instance-location-id"),
resource_version=service_instance.get("resource-version"),
selflink=service_instance.get("selflink"),
orchestration_status=service_instance.get("orchestration-status"),
input_parameters=service_instance.get("input-parameters")
)
@classmethod
def get_all_url(cls, customer: "Customer") -> str: # pylint: disable=arguments-differ
"""Return url to get all customers.
Returns:
str: Url to get all customers
"""
return (f"{cls.base_url}{cls.api_version}/business/customers/"
f"customer/{customer.global_customer_id}/service-subscriptions/")
@classmethod
def create_from_api_response(cls,
api_response: dict,
customer: "Customer") -> "ServiceSubscription":
"""Create service subscription using API response dict.
Returns:
ServiceSubscription: ServiceSubscription object.
"""
return cls(
service_type=api_response.get("service-type"),
resource_version=api_response.get("resource-version"),
customer=customer
)
@property
def url(self) -> str:
"""Cloud region object url.
URL used to call CloudRegion A&AI API
Returns:
str: CloudRegion object url
"""
return (
f"{self.base_url}{self.api_version}/business/customers/"
f"customer/{self.customer.global_customer_id}/service-subscriptions/"
f"service-subscription/{self.service_type}"
)
@property
def service_instances(self) -> Iterator[ServiceInstance]:
"""Service instances.
Yields:
Iterator[ServiceInstance]: Service instance
"""
for service_instance in \
self.send_message_json("GET",
(f"Get all service instances for {self.service_type} service "
f"subscription"),
f"{self.url}/service-instances").get("service-instance", []):
yield ServiceInstance(
service_subscription=self,
instance_id=service_instance.get("service-instance-id"),
instance_name=service_instance.get("service-instance-name"),
service_type=service_instance.get("service-type"),
service_role=service_instance.get("service-role"),
environment_context=service_instance.get("environment-context"),
workload_context=service_instance.get("workload-context"),
created_at=service_instance.get("created-at"),
updated_at=service_instance.get("updated-at"),
description=service_instance.get("description"),
model_invariant_id=service_instance.get("model-invariant-id"),
model_version_id=service_instance.get("model-version-id"),
persona_model_version=service_instance.get("persona-model-version"),
widget_model_id=service_instance.get("widget-model-id"),
widget_model_version=service_instance.get("widget-model-version"),
bandwith_total=service_instance.get("bandwidth-total"),
vhn_portal_url=service_instance.get("vhn-portal-url"),
service_instance_location_id=service_instance.get("service-instance-location-id"),
resource_version=service_instance.get("resource-version"),
selflink=service_instance.get("selflink"),
orchestration_status=service_instance.get("orchestration-status"),
input_parameters=service_instance.get("input-parameters")
)
@property
def tenant_relationships(self) -> Iterator["Relationship"]:
"""Tenant related relationships.
Iterate through relationships and get related to tenant.
Yields:
Relationship: Relationship related to tenant.
"""
for relationship in self.relationships:
if relationship.related_to == "tenant":
yield relationship
@property
def cloud_region(self) -> "CloudRegion":
"""Cloud region associated with service subscription.
IT'S DEPRECATED! `cloud_regions` parameter SHOULD BE USED
Raises:
ParameterError: Service subscription has no associated cloud region.
Returns:
CloudRegion: CloudRegion object
"""
try:
return next(self.cloud_regions)
except StopIteration:
msg = f"No cloud region for service subscription '{self.name}'"
raise ParameterError(msg)
@property
def tenant(self) -> "Tenant":
"""Tenant associated with service subscription.
IT'S DEPRECATED! `tenants` parameter SHOULD BE USED
Raises:
ParameterError: Service subscription has no associated tenants
Returns:
Tenant: Tenant object
"""
try:
return next(self.tenants)
except StopIteration:
msg = f"No tenants for service subscription '{self.name}'"
raise ParameterError(msg)
@property
def _cloud_regions_tenants_data(self) -> Iterator["ServiceSubscriptionCloudRegionTenantData"]:
for relationship in self.tenant_relationships:
cr_tenant_data: ServiceSubscriptionCloudRegionTenantData = \
ServiceSubscriptionCloudRegionTenantData()
for data in relationship.relationship_data:
if data["relationship-key"] == "cloud-region.cloud-owner":
cr_tenant_data.cloud_owner = data["relationship-value"]
if data["relationship-key"] == "cloud-region.cloud-region-id":
cr_tenant_data.cloud_region_id = data["relationship-value"]
if data["relationship-key"] == "tenant.tenant-id":
cr_tenant_data.tenant_id = data["relationship-value"]
if all([cr_tenant_data.cloud_owner,
cr_tenant_data.cloud_region_id,
cr_tenant_data.tenant_id]):
yield cr_tenant_data
else:
self._logger.error("Invalid tenant relationship: %s", relationship)
@property
def cloud_regions(self) -> Iterator["CloudRegion"]:
"""Cloud regions associated with service subscription.
Yields:
CloudRegion: CloudRegion object
"""
cloud_region_set: set = set()
for cr_data in self._cloud_regions_tenants_data:
cloud_region_set.add((cr_data.cloud_owner, cr_data.cloud_region_id))
for cloud_region_data in cloud_region_set:
try:
yield CloudRegion.get_by_id(cloud_owner=cloud_region_data[0],
cloud_region_id=cloud_region_data[1])
except ResourceNotFound:
self._logger.error("Can't get cloud region %s %s", cloud_region_data[0], \
cloud_region_data[1])
@property
def tenants(self) -> Iterator["Tenant"]:
"""Tenants associated with service subscription.
Yields:
Tenant: Tenant object
"""
for cr_data in self._cloud_regions_tenants_data:
try:
cloud_region: CloudRegion = CloudRegion.get_by_id(cr_data.cloud_owner,
cr_data.cloud_region_id)
yield cloud_region.get_tenant(cr_data.tenant_id)
except ResourceNotFound:
self._logger.error("Can't get %s tenant", cr_data.tenant_id)
def get_service_instance_by_id(self, service_instance_id) -> ServiceInstance:
"""Get service instance using it's ID.
Args:
service_instance_id (str): ID of the service instance
Returns:
ServiceInstance: ServiceInstance object
"""
return self._get_service_instance_by_filter_parameter(
"service-instance-id",
service_instance_id
)
def get_service_instance_by_name(self, service_instance_name: str) -> ServiceInstance:
"""Get service instance using it's name.
Args:
service_instance_name (str): Name of the service instance
Returns:
ServiceInstance: ServiceInstance object
"""
return self._get_service_instance_by_filter_parameter(
"service-instance-name",
service_instance_name
)
def link_to_cloud_region_and_tenant(self,
cloud_region: "CloudRegion",
tenant: "Tenant") -> None:
"""Create relationship between object and cloud region with tenant.
Args:
cloud_region (CloudRegion): Cloud region to link to
tenant (Tenant): Cloud region tenant to link to
"""
relationship: Relationship = Relationship(
related_to="tenant",
related_link=tenant.url,
relationship_data=[
{
"relationship-key": "cloud-region.cloud-owner",
"relationship-value": cloud_region.cloud_owner,
},
{
"relationship-key": "cloud-region.cloud-region-id",
"relationship-value": cloud_region.cloud_region_id,
},
{
"relationship-key": "tenant.tenant-id",
"relationship-value": tenant.tenant_id,
},
],
related_to_property=[
{"property-key": "tenant.tenant-name", "property-value": tenant.name}
],
)
self.add_relationship(relationship)
class Customer(AaiResource):
"""Customer class."""
def __init__(self,
global_customer_id: str,
subscriber_name: str,
subscriber_type: str,
resource_version: str = None) -> None:
"""Initialize Customer class object.
Args:
global_customer_id (str): Global customer id used across ONAP to
uniquely identify customer.
subscriber_name (str): Subscriber name, an alternate way to retrieve a customer.
subscriber_type (str): Subscriber type, a way to provide VID with
only the INFRA customers.
resource_version (str, optional): Used for optimistic concurrency.
Must be empty on create, valid on update
and delete. Defaults to None.
"""
super().__init__()
self.global_customer_id: str = global_customer_id
self.subscriber_name: str = subscriber_name
self.subscriber_type: str = subscriber_type
self.resource_version: str = resource_version
def __repr__(self) -> str: # noqa
"""Customer description.
Returns:
str: Customer object description
"""
return (f"Customer(global_customer_id={self.global_customer_id}, "
f"subscriber_name={self.subscriber_name}, "
f"subscriber_type={self.subscriber_type}, "
f"resource_version={self.resource_version})")
def get_service_subscription_by_service_type(self, service_type: str) -> ServiceSubscription:
"""Get subscribed service by service type.
Call a request to get service subscriptions filtered by service-type parameter.
Args:
service_type (str): Service type
Returns:
ServiceSubscription: Service subscription
"""
response: dict = self.send_message_json(
"GET",
f"Get service subscription with {service_type} service type",
(f"{self.base_url}{self.api_version}/business/customers/"
f"customer/{self.global_customer_id}/service-subscriptions"
f"?service-type={service_type}")
)
return ServiceSubscription.create_from_api_response(response["service-subscription"][0],
self)
@classmethod
def get_all_url(cls) -> str: # pylint: disable=arguments-differ
"""Return an url to get all customers.
Returns:
str: URL to get all customers
"""
return f"{cls.base_url}{cls.api_version}/business/customers"
@classmethod
def get_all(cls,
global_customer_id: str = None,
subscriber_name: str = None,
subscriber_type: str = None) -> Iterator["Customer"]:
"""Get all customers.
Call an API to retrieve all customers. It can be filtered
by global-customer-id, subscriber-name and/or subsriber-type.
Args:
global_customer_id (str): global-customer-id to filer customers by. Defaults to None.
subscriber_name (str): subscriber-name to filter customers by. Defaults to None.
subscriber_type (str): subscriber-type to filter customers by. Defaults to None.
"""
filter_parameters: dict = cls.filter_none_key_values(
{
"global-customer-id": global_customer_id,
"subscriber-name": subscriber_name,
"subscriber-type": subscriber_type,
}
)
url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}")
for customer in cls.send_message_json("GET", "get customers", url).get("customer", []):
yield Customer(
global_customer_id=customer["global-customer-id"],
subscriber_name=customer["subscriber-name"],
subscriber_type=customer["subscriber-type"],
resource_version=customer["resource-version"],
)
@classmethod
def get_by_global_customer_id(cls, global_customer_id: str) -> "Customer":
"""Get customer by it's global customer id.
Args:
global_customer_id (str): global customer ID
Returns:
Customer: Customer with given global_customer_id
"""
response: dict = cls.send_message_json(
"GET",
f"Get {global_customer_id} customer",
f"{cls.base_url}{cls.api_version}/business/customers/customer/{global_customer_id}"
)
return Customer(
global_customer_id=response["global-customer-id"],
subscriber_name=response["subscriber-name"],
subscriber_type=response["subscriber-type"],
resource_version=response["resource-version"],
)
@classmethod
def create(cls,
global_customer_id: str,
subscriber_name: str,
subscriber_type: str,
service_subscriptions: Optional[Iterable[str]] = None) -> "Customer":
"""Create customer.
Args:
global_customer_id (str): Global customer id used across ONAP
to uniquely identify customer.
subscriber_name (str): Subscriber name, an alternate way
to retrieve a customer.
subscriber_type (str): Subscriber type, a way to provide
VID with only the INFRA customers.
service_subscriptions (Optional[Iterable[str]], optional): Iterable
of service subscription names should be created for newly
created customer. Defaults to None.
Returns:
Customer: Customer object.
"""
url: str = (
f"{cls.base_url}{cls.api_version}/business/customers/"
f"customer/{global_customer_id}"
)
cls.send_message(
"PUT",
"declare customer",
url,
data=jinja_env()
.get_template("customer_create.json.j2")
.render(
global_customer_id=global_customer_id,
subscriber_name=subscriber_name,
subscriber_type=subscriber_type,
service_subscriptions=service_subscriptions
),
)
response: dict = cls.send_message_json(
"GET", "get created customer", url
) # Call API one more time to get Customer's resource version
return Customer(
global_customer_id=response["global-customer-id"],
subscriber_name=response["subscriber-name"],
subscriber_type=response["subscriber-type"],
resource_version=response["resource-version"],
)
@property
def url(self) -> str:
"""Return customer object url.
Unique url address to get customer's data.
Returns:
str: Customer object url
"""
return (
f"{self.base_url}{self.api_version}/business/customers/customer/"
f"{self.global_customer_id}?resource-version={self.resource_version}"
)
@property
def service_subscriptions(self) -> Iterator[ServiceSubscription]:
"""Service subscriptions of customer resource.
Yields:
ServiceSubscription: ServiceSubscription object
"""
try:
response: dict = self.send_message_json(
"GET",
"get customer service subscriptions",
f"{self.base_url}{self.api_version}/business/customers/"
f"customer/{self.global_customer_id}/service-subscriptions"
)
for service_subscription in response.get("service-subscription", []):
yield ServiceSubscription.create_from_api_response(
service_subscription,
self
)
except ResourceNotFound as exc:
self._logger.info(
"Subscriptions are not " \
"found for a customer: %s", exc)
except APIError as exc:
self._logger.error(
"API returned an error: %s", exc)
def subscribe_service(self, service_type: str) -> "ServiceSubscription":
"""Create SDC Service subscription.
If service subscription with given service_type already exists it won't create
a new resource but use the existing one.
Args:
service_type (str): Value defined by orchestration to identify this service
across ONAP.
"""
try:
return self.get_service_subscription_by_service_type(service_type)
except ResourceNotFound:
self._logger.info("Create service subscription for %s customer",
self.global_customer_id)
self.send_message(
"PUT",
"Create service subscription",
(f"{self.base_url}{self.api_version}/business/customers/"
f"customer/{self.global_customer_id}/service-subscriptions/"
f"service-subscription/{service_type}")
)
return self.get_service_subscription_by_service_type(service_type)
def delete(self) -> None:
"""Delete customer.
Sends request to A&AI to delete customer object.
"""
self.send_message(
"DELETE",
"Delete customer",
self.url
)