diff options
Diffstat (limited to 'src/onapsdk/aai/cloud_infrastructure')
| -rw-r--r-- | src/onapsdk/aai/cloud_infrastructure/__init__.py | 18 | ||||
| -rw-r--r-- | src/onapsdk/aai/cloud_infrastructure/cloud_region.py | 621 | ||||
| -rw-r--r-- | src/onapsdk/aai/cloud_infrastructure/complex.py | 300 | ||||
| -rw-r--r-- | src/onapsdk/aai/cloud_infrastructure/geo_region.py | 191 | ||||
| -rw-r--r-- | src/onapsdk/aai/cloud_infrastructure/tenant.py | 101 |
5 files changed, 1231 insertions, 0 deletions
diff --git a/src/onapsdk/aai/cloud_infrastructure/__init__.py b/src/onapsdk/aai/cloud_infrastructure/__init__.py new file mode 100644 index 0000000..a380ce3 --- /dev/null +++ b/src/onapsdk/aai/cloud_infrastructure/__init__.py @@ -0,0 +1,18 @@ +"""A&AI cloud infrastructure package.""" +# 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 .cloud_region import AvailabilityZone, CloudRegion, EsrSystemInfo +from .complex import Complex +from .geo_region import GeoRegion +from .tenant import Tenant diff --git a/src/onapsdk/aai/cloud_infrastructure/cloud_region.py b/src/onapsdk/aai/cloud_infrastructure/cloud_region.py new file mode 100644 index 0000000..d57a025 --- /dev/null +++ b/src/onapsdk/aai/cloud_infrastructure/cloud_region.py @@ -0,0 +1,621 @@ +"""Cloud region 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 dataclasses import dataclass +from typing import Any, Dict, Iterator, List, Optional +from urllib.parse import urlencode + +from onapsdk.msb.multicloud import Multicloud +from onapsdk.utils.jinja import jinja_env +from onapsdk.exceptions import ResourceNotFound + +from ..aai_element import AaiResource, Relationship +from .complex import Complex +from .tenant import Tenant + + +@dataclass +class AvailabilityZone: + """Availability zone. + + A collection of compute hosts/pservers + """ + + name: str + hypervisor_type: str + operational_status: str = None + resource_version: str = None + + +@dataclass +class EsrSystemInfo: # pylint: disable=too-many-instance-attributes + """Persist common address information of external systems.""" + + esr_system_info_id: str + user_name: str + password: str + system_type: str + resource_version: str + system_name: str = None + esr_type: str = None + vendor: str = None + version: str = None + service_url: str = None + protocol: str = None + ssl_cacert: str = None + ssl_insecure: Optional[bool] = None + ip_address: str = None + port: str = None + cloud_domain: str = None + default_tenant: str = None + passive: Optional[bool] = None + remote_path: str = None + system_status: str = None + openstack_region_id: str = None + + +class CloudRegion(AaiResource): # pylint: disable=too-many-instance-attributes + """Cloud region class. + + Represents A&AI cloud region object. + """ + + def __init__(self, + cloud_owner: str, + cloud_region_id: str, + orchestration_disabled: bool, + in_maint: bool, + *, # rest of parameters are keyword + cloud_type: str = "", + owner_defined_type: str = "", + cloud_region_version: str = "", + identity_url: str = "", + cloud_zone: str = "", + complex_name: str = "", + sriov_automation: str = "", + cloud_extra_info: str = "", + upgrade_cycle: str = "", + resource_version: str = "") -> None: + """Cloud region object initialization. + + Args: + cloud_owner (str): Identifies the vendor and cloud name. + cloud_region_id (str): Identifier used by the vendor for the region. + orchestration_disabled (bool): Used to indicate whether orchestration is + enabled for this cloud-region. + in_maint (bool): Used to indicate whether or not cloud-region object + is in maintenance mode. + owner_defined_type (str, optional): Cloud-owner defined type + indicator (e.g., dcp, lcp). Defaults to "". + cloud_region_version (str, optional): Software version employed at the site. + Defaults to "". + identity_url (str, optional): URL of the keystone identity service. Defaults to "". + cloud_zone (str, optional): Zone where the cloud is homed. Defaults to "". + complex_name (str, optional): Complex name for cloud-region instance. Defaults to "". + sriov_automation (str, optional): Whether the cloud region supports (true) or does + not support (false) SR-IOV automation. Defaults to "". + cloud_extra_info (str, optional): ESR inputs extra information about the VIM or Cloud + which will be decoded by MultiVIM. Defaults to "". + upgrade_cycle (str, optional): Upgrade cycle for the cloud region. + For AIC regions upgrade cycle is designated by A,B,C etc. Defaults to "". + resource_version (str, optional): Used for optimistic concurrency. + Must be empty on create, valid on update and delete. Defaults to "". + + """ + super().__init__() + self.cloud_owner = cloud_owner + self.cloud_region_id = cloud_region_id + self.orchestration_disabled = orchestration_disabled + self.in_maint = in_maint + self.cloud_type = cloud_type + self.owner_defined_type = owner_defined_type + self.cloud_region_version = cloud_region_version + self.identity_url = identity_url + self.cloud_zone = cloud_zone + self.complex_name = complex_name + self.sriov_automation = sriov_automation + self.cloud_extra_info = cloud_extra_info + self.upgrade_cycle = upgrade_cycle + self.resource_version = resource_version + + def __repr__(self) -> str: + """Cloud region object representation. + + Returns: + str: Human readable string contains most important information about cloud region. + + """ + return ( + f"CloudRegion(cloud_owner={self.cloud_owner}, cloud_region_id={self.cloud_region_id})" + ) + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return url to get all cloud regions. + + Returns: + str: Url to get all cloud regions + + """ + return f"{cls.base_url}{cls.api_version}/cloud-infrastructure/cloud-regions" + + @classmethod + def get_all(cls, + cloud_owner: str = None, + cloud_region_id: str = None, + cloud_type: str = None, + owner_defined_type: str = None) -> Iterator["CloudRegion"]: + """Get all A&AI cloud regions. + + Cloud regions can be filtered by 4 parameters: cloud-owner, + cloud-region-id, cloud-type and owner-defined-type. + + Yields: + CloudRegion -- CloudRegion object. Can not yield anything + if cloud region with given filter parameters doesn't exist + + """ + # Filter request parameters - use only these which are not None + filter_parameters: dict = cls.filter_none_key_values( + { + "cloud-owner": cloud_owner, + "cloud-region-id": cloud_region_id, + "cloud-type": cloud_type, + "owner-defined-type": owner_defined_type, + } + ) + url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}") + response_json: Dict[str, List[Dict[str, Any]]] = cls.send_message_json( + "GET", "get cloud regions", url + ) + for cloud_region in response_json.get("cloud-region", []): # typing: dict + yield CloudRegion( + cloud_owner=cloud_region["cloud-owner"], # required + cloud_region_id=cloud_region["cloud-region-id"], # required + cloud_type=cloud_region.get("cloud-type"), + owner_defined_type=cloud_region.get("owner-defined-type"), + cloud_region_version=cloud_region.get("cloud-region-version"), + identity_url=cloud_region.get("identity_url"), + cloud_zone=cloud_region.get("cloud-zone"), + complex_name=cloud_region.get("complex-name"), + sriov_automation=cloud_region.get("sriov-automation"), + cloud_extra_info=cloud_region.get("cloud-extra-info"), + upgrade_cycle=cloud_region.get("upgrade-cycle"), + orchestration_disabled=cloud_region["orchestration-disabled"], # required + in_maint=cloud_region["in-maint"], # required + resource_version=cloud_region.get("resource-version"), + ) + + @classmethod + def get_by_id(cls, cloud_owner: str, cloud_region_id: str) -> "CloudRegion": + """Get CloudRegion object by cloud_owner and cloud-region-id field value. + + This method calls A&AI cloud region API filtering them by cloud_owner and + cloud-region-id field value. + + Raises: + ResourceNotFound: Cloud region with given id does not exist. + + Returns: + CloudRegion: CloudRegion object with given cloud-region-id. + + """ + try: + return next(cls.get_all(cloud_owner=cloud_owner, cloud_region_id=cloud_region_id)) + except StopIteration: + msg = ( + f'CloudRegion with {cloud_owner}, ' + f'{cloud_region_id} cloud-id not found. ' + ) + raise ResourceNotFound(msg) + + @classmethod + def create(cls, # pylint: disable=too-many-locals + cloud_owner: str, + cloud_region_id: str, + orchestration_disabled: bool, + in_maint: bool, + *, # rest of parameters are keyword + cloud_type: str = "", + owner_defined_type: str = "", + cloud_region_version: str = "", + identity_url: str = "", + cloud_zone: str = "", + complex_name: str = "", + sriov_automation: str = "", + cloud_extra_info: str = "", + upgrade_cycle: str = "") -> "CloudRegion": + """Create CloudRegion object. + + Create cloud region with given values. + + Returns: + CloudRegion: Created cloud region. + + """ + cloud_region: "CloudRegion" = CloudRegion( + cloud_owner=cloud_owner, + cloud_region_id=cloud_region_id, + orchestration_disabled=orchestration_disabled, + in_maint=in_maint, + cloud_type=cloud_type, + owner_defined_type=owner_defined_type, + cloud_region_version=cloud_region_version, + identity_url=identity_url, + cloud_zone=cloud_zone, + complex_name=complex_name, + sriov_automation=sriov_automation, + cloud_extra_info=cloud_extra_info, + upgrade_cycle=upgrade_cycle, + ) + url: str = ( + f"{cls.base_url}{cls.api_version}/cloud-infrastructure/cloud-regions/cloud-region/" + f"{cloud_region.cloud_owner}/{cloud_region.cloud_region_id}" + ) + cls.send_message( + "PUT", + "Create cloud region", + url, + data=jinja_env() + .get_template("cloud_region_create.json.j2") + .render(cloud_region=cloud_region), + ) + return cloud_region + + @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}/cloud-infrastructure/cloud-regions/cloud-region/" + f"{self.cloud_owner}/{self.cloud_region_id}" + ) + + @property + def tenants(self) -> Iterator["Tenant"]: + """Tenants iterator. + + Cloud region tenants iterator. + + Returns: + Iterator[Tenant]: Iterate through cloud region tenants + + """ + response: dict = self.send_message_json("GET", "get tenants", f"{self.url}/tenants") + return ( + Tenant( + cloud_region=self, + tenant_id=tenant["tenant-id"], + tenant_name=tenant["tenant-name"], + tenant_context=tenant.get("tenant-context"), + resource_version=tenant.get("resource-version"), + ) + for tenant in response.get("tenant", []) + ) + + @property + def availability_zones(self) -> Iterator[AvailabilityZone]: + """Cloud region availability zones. + + Iterate over CloudRegion availability zones. Relationship list is given using A&AI API call. + + Returns: + Iterator[AvailabilityZone]: CloudRegion availability zone + + """ + response: dict = self.send_message_json( + "GET", "get cloud region availability zones", f"{self.url}/availability-zones" + ) + return ( + AvailabilityZone( + name=availability_zone["availability-zone-name"], + hypervisor_type=availability_zone["hypervisor-type"], + operational_status=availability_zone.get("operational-status"), + resource_version=availability_zone.get("resource-version") + ) + for availability_zone in response.get("availability-zone", []) + ) + + @property + def esr_system_infos(self) -> Iterator[EsrSystemInfo]: + """Cloud region collection of persistent block-level external system auth info. + + Returns: + Iterator[EsrSystemInfo]: Cloud region external system address information. + + """ + response: dict = self.send_message_json( + "GET", "get cloud region external systems info list", f"{self.url}/esr-system-info-list" + ) + return ( + EsrSystemInfo( + esr_system_info_id=esr_system_info.get("esr-system-info-id"), + user_name=esr_system_info.get("user-name"), + password=esr_system_info.get("password"), + system_type=esr_system_info.get("system-type"), + system_name=esr_system_info.get("system-name"), + esr_type=esr_system_info.get("type"), + vendor=esr_system_info.get("vendor"), + version=esr_system_info.get("version"), + service_url=esr_system_info.get("service-url"), + protocol=esr_system_info.get("protocol"), + ssl_cacert=esr_system_info.get("ssl-cacert"), + ssl_insecure=esr_system_info.get("ssl-insecure"), + ip_address=esr_system_info.get("ip-address"), + port=esr_system_info.get("port"), + cloud_domain=esr_system_info.get("cloud-domain"), + default_tenant=esr_system_info.get("default-tenant"), + passive=esr_system_info.get("passive"), + remote_path=esr_system_info.get("remote-path"), + system_status=esr_system_info.get("system-status"), + openstack_region_id=esr_system_info.get("openstack-region-id"), + resource_version=esr_system_info.get("resource-version"), + ) + for esr_system_info in response.get("esr-system-info", []) + ) + + @property + def complex(self) -> Optional[Complex]: + """Complex related with cloud region. + + Returns: + Optional[Complex]: Complex object related with CloudRegion or None if + CloudRegion has no relationship with any Complex + + """ + try: + for relationship in self.relationships: + if relationship.related_to == "complex": + physical_location_id: Optional[str] = relationship.get_relationship_data( + "complex.physical-location-id" + ) + if physical_location_id is not None: + try: + return Complex.get_by_physical_location_id( + physical_location_id + ) + except ResourceNotFound: + self._logger.error("Complex with %s physical location id does " + "not exist", physical_location_id) + self._logger.error("Invalid Complex relationship!") + return None + except ResourceNotFound: + self._logger.debug("Cloud region %s has no relationships", self.cloud_region_id) + self._logger.debug("Cloud region %s has no related complex", self.cloud_region_id) + return None + + def add_tenant(self, tenant_id: str, tenant_name: str, tenant_context: str = None) -> None: + """Add tenant to cloud region. + + Args: + tenant_id (str): Unique id relative to the cloud-region. + tenant_name (str): Readable name of tenant + tenant_context (str, optional): This field will store + the tenant context.. Defaults to None. + + """ + self.send_message( + "PUT", + "add tenant to cloud region", + f"{self.url}/tenants/tenant/{tenant_id}", + data=jinja_env() + .get_template("cloud_region_add_tenant.json.j2") + .render(tenant_id=tenant_id, tenant_name=tenant_name, + tenant_context=tenant_context) + ) + + def get_tenant(self, tenant_id: str) -> "Tenant": + """Get tenant with provided ID. + + Args: + tenant_id (str): Tenant ID + + Returns: + Tenant: Tenant object + + """ + response: dict = self.send_message_json( + "GET", + "get tenants", + f"{self.url}/tenants/tenant/{tenant_id}" + ) + return Tenant( + cloud_region=self, + tenant_id=response["tenant-id"], + tenant_name=response["tenant-name"], + tenant_context=response.get("tenant-context"), + resource_version=response.get("resource-version"), + ) + + def get_tenants_by_name(self, tenant_name: str) -> Iterator["Tenant"]: + """Get tenants with given name. + + Args: + tenant_name (str): Tenant name + + Returns: + Iterator[Tenant]: Iterate through cloud region tenants with given name + + """ + return (tenant for tenant in self.tenants if tenant.name == tenant_name) + + + def get_availability_zone_by_name(self, + zone_name: str) -> "AvailabilityZone": + """Get availability zone with provided Name. + + Args: + availability_zone name (str): The name of the availibilty zone + + Returns: + AvailabilityZone: AvailabilityZone object + + """ + response: dict = self.send_message_json( + "GET", + "get availability_zones", + f"{self.url}/availability-zones/availability-zone/{zone_name}" + ) + return AvailabilityZone( + name=response["availability-zone-name"], + hypervisor_type=response["hypervisor-type"], + resource_version=response["resource-version"] + ) + + def add_availability_zone(self, + availability_zone_name: str, + availability_zone_hypervisor_type: str, + availability_zone_operational_status: str = None) -> None: + """Add avaiability zone to cloud region. + + Args: + availability_zone_name (str): Name of the availability zone. + Unique across a cloud region + availability_zone_hypervisor_type (str): Type of hypervisor + availability_zone_operational_status (str, optional): State that indicates whether + the availability zone should be used. Defaults to None. + """ + self.send_message( + "PUT", + "Add availability zone to cloud region", + f"{self.url}/availability-zones/availability-zone/{availability_zone_name}", + data=jinja_env() + .get_template("cloud_region_add_availability_zone.json.j2") + .render(availability_zone_name=availability_zone_name, + availability_zone_hypervisor_type=availability_zone_hypervisor_type, + availability_zone_operational_status=availability_zone_operational_status) + ) + + def add_esr_system_info(self, # pylint: disable=too-many-arguments, too-many-locals + esr_system_info_id: str, + user_name: str, + password: str, + system_type: str, + system_name: str = None, + esr_type: str = None, + vendor: str = None, + version: str = None, + service_url: str = None, + protocol: str = None, + ssl_cacert: str = None, + ssl_insecure: Optional[bool] = None, + ip_address: str = None, + port: str = None, + cloud_domain: str = None, + default_tenant: str = None, + passive: Optional[bool] = None, + remote_path: str = None, + system_status: str = None, + openstack_region_id: str = None, + resource_version: str = None) -> None: + """Add external system info to cloud region. + + Args: + esr_system_info_id (str): Unique ID of esr system info + user_name (str): username used to access external system + password (str): password used to access external system + system_type (str): it could be vim/vnfm/thirdparty-sdnc/ + ems-resource/ems-performance/ems-alarm + system_name (str, optional): name of external system. Defaults to None. + esr_type (str, optional): type of external system. Defaults to None. + vendor (str, optional): vendor of external system. Defaults to None. + version (str, optional): version of external system. Defaults to None. + service_url (str, optional): url used to access external system. Defaults to None. + protocol (str, optional): protocol of third party SDNC, + for example netconf/snmp. Defaults to None. + ssl_cacert (str, optional): ca file content if enabled ssl on auth-url. + Defaults to None. + ssl_insecure (bool, optional): Whether to verify VIM's certificate. Defaults to True. + ip_address (str, optional): service IP of ftp server. Defaults to None. + port (str, optional): service port of ftp server. Defaults to None. + cloud_domain (str, optional): domain info for authentication. Defaults to None. + default_tenant (str, optional): default tenant of VIM. Defaults to None. + passive (bool, optional): ftp passive mode or not. Defaults to False. + remote_path (str, optional): resource or performance data file path. Defaults to None. + system_status (str, optional): he status of external system. Defaults to None. + openstack_region_id (str, optional): OpenStack region ID used by MultiCloud plugin to + interact with an OpenStack instance. Defaults to None. + """ + self.send_message( + "PUT", + "Add external system info to cloud region", + f"{self.url}/esr-system-info-list/esr-system-info/{esr_system_info_id}", + data=jinja_env() + .get_template("cloud_region_add_esr_system_info.json.j2") + .render(esr_system_info_id=esr_system_info_id, + user_name=user_name, + password=password, + system_type=system_type, + system_name=system_name, + esr_type=esr_type, + vendor=vendor, + version=version, + service_url=service_url, + protocol=protocol, + ssl_cacert=ssl_cacert, + ssl_insecure=ssl_insecure, + ip_address=ip_address, + port=port, + cloud_domain=cloud_domain, + default_tenant=default_tenant, + passive=passive, + remote_path=remote_path, + system_status=system_status, + openstack_region_id=openstack_region_id, + resource_version=resource_version) + ) + + def register_to_multicloud(self, default_tenant: str = None) -> None: + """Register cloud to multicloud using MSB API. + + Args: + default_tenant (str, optional): Default tenant. Defaults to None. + """ + Multicloud.register_vim(self.cloud_owner, self.cloud_region_id, default_tenant) + + def unregister_from_multicloud(self) -> None: + """Unregister cloud from mutlicloud.""" + Multicloud.unregister_vim(self.cloud_owner, self.cloud_region_id) + + def delete(self) -> None: + """Delete cloud region.""" + self.send_message( + "DELETE", + f"Delete cloud region {self.cloud_region_id}", + self.url, + params={"resource-version": self.resource_version} + ) + + def link_to_complex(self, complex_object: Complex) -> None: + """Link cloud region to comples. + + It creates relationhip object and add it into cloud region. + """ + relationship = Relationship( + related_to="complex", + related_link=(f"aai/v13/cloud-infrastructure/complexes/" + f"complex/{complex_object.physical_location_id}"), + relationship_data={ + "relationship-key": "complex.physical-location-id", + "relationship-value": f"{complex_object.physical_location_id}", + }, + relationship_label="org.onap.relationships.inventory.LocatedIn", + ) + self.add_relationship(relationship) diff --git a/src/onapsdk/aai/cloud_infrastructure/complex.py b/src/onapsdk/aai/cloud_infrastructure/complex.py new file mode 100644 index 0000000..a854f02 --- /dev/null +++ b/src/onapsdk/aai/cloud_infrastructure/complex.py @@ -0,0 +1,300 @@ +"""A&AI Complex 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 Any, Dict, Iterator +from urllib.parse import urlencode + +from onapsdk.utils.jinja import jinja_env + +from ..aai_element import AaiResource + + +class Complex(AaiResource): # pylint: disable=too-many-instance-attributes + """Complex class. + + Collection of physical locations that can house cloud-regions. + """ + + def __init__(self, # pylint: disable=too-many-locals + physical_location_id: str, + *, + name: str = "", + data_center_code: str = "", + identity_url: str = "", + resource_version: str = "", + physical_location_type: str = "", + street1: str = "", + street2: str = "", + city: str = "", + state: str = "", + postal_code: str = "", + country: str = "", + region: str = "", + latitude: str = "", + longitude: str = "", + elevation: str = "", + lata: str = "", + timezone: str = "", + data_owner: str = "", + data_source: str = "", + data_source_version: str = "") -> None: + """Complex object initialization. + + Args: + name (str): complex name + physical_location_id (str): complex ID + data_center_code (str, optional): complex data center code. Defaults to "". + identity_url (str, optional): complex identity url. Defaults to "". + resource_version (str, optional): complex resource version. Defaults to "". + physical_location_type (str, optional): complex physical location type. Defaults to "". + street1 (str, optional): complex address street part one. Defaults to "". + street2 (str, optional): complex address street part two. Defaults to "". + city (str, optional): complex address city. Defaults to "". + state (str, optional): complex address state. Defaults to "". + postal_code (str, optional): complex address postal code. Defaults to "". + country (str, optional): complex address country. Defaults to "". + region (str, optional): complex address region. Defaults to "". + latitude (str, optional): complex geographical location latitude. Defaults to "". + longitude (str, optional): complex geographical location longitude. Defaults to "". + elevation (str, optional): complex elevation. Defaults to "". + lata (str, optional): complex lata. Defaults to "". + timezone (str, optional): the time zone where the complex is located. Defaults to "". + data_owner (str, optional): Identifies the entity that is responsible managing this + inventory object. Defaults to "". + data_source (str, optional): Identifies the upstream source of the data. Defaults to "". + data_source_version (str, optional): Identifies the version of the upstream source. + Defaults to "". + + """ + super().__init__() + self.name: str = name + self.physical_location_id: str = physical_location_id + self.data_center_code: str = data_center_code + self.identity_url: str = identity_url + self.resource_version: str = resource_version + self.physical_location_type: str = physical_location_type + self.street1: str = street1 + self.street2: str = street2 + self.city: str = city + self.state: str = state + self.postal_code: str = postal_code + self.country: str = country + self.region: str = region + self.latitude: str = latitude + self.longitude: str = longitude + self.elevation: str = elevation + self.lata: str = lata + self.timezone: str = timezone + self.data_owner: str = data_owner + self.data_source: str = data_source + self.data_source_version: str = data_source_version + + def __repr__(self) -> str: + """Complex object description. + + Returns: + str: Complex object description + + """ + return (f"Complex(name={self.name}, " + f"physical_location_id={self.physical_location_id}, " + f"resource_version={self.resource_version})") + + @property + def url(self) -> str: + """Complex url. + + Returns: + str: Complex url + + """ + return (f"{self.base_url}{self.api_version}/cloud-infrastructure/complexes/complex/" + f"{self.physical_location_id}") + + @classmethod + def create(cls, # pylint: disable=too-many-locals + physical_location_id: str, + *, + name: str = "", + data_center_code: str = "", + identity_url: str = "", + resource_version: str = "", + physical_location_type: str = "", + street1: str = "", + street2: str = "", + city: str = "", + state: str = "", + postal_code: str = "", + country: str = "", + region: str = "", + latitude: str = "", + longitude: str = "", + elevation: str = "", + lata: str = "", + timezone: str = "", + data_owner: str = "", + data_source: str = "", + data_source_version: str = "") -> "Complex": + """Create complex. + + Create complex object by calling A&AI API. + If API request doesn't fail it returns Complex object. + + Returns: + Complex: Created complex object + + """ + complex_object: Complex = Complex( + name=name, + physical_location_id=physical_location_id, + data_center_code=data_center_code, + identity_url=identity_url, + resource_version=resource_version, + physical_location_type=physical_location_type, + street1=street1, + street2=street2, + city=city, + state=state, + postal_code=postal_code, + country=country, + region=region, + latitude=latitude, + longitude=longitude, + elevation=elevation, + lata=lata, + timezone=timezone, + data_owner=data_owner, + data_source=data_source, + data_source_version=data_source_version + ) + payload: str = jinja_env().get_template("complex_create.json.j2").render( + complex=complex_object) + url: str = ( + f"{cls.base_url}{cls.api_version}/cloud-infrastructure/complexes/complex/" + f"{complex_object.physical_location_id}" + ) + cls.send_message("PUT", "create complex", url, data=payload) + return complex_object + + @classmethod + def get_all_url(cls) -> str: # pylint: disable=arguments-differ + """Return an url to get all complexes. + + Returns: + str: URL to get all complexes + + """ + return f"{cls.base_url}{cls.api_version}/cloud-infrastructure/complexes" + + @classmethod + def get_all(cls, + physical_location_id: str = None, + data_center_code: str = None, + complex_name: str = None, + identity_url: str = None) -> Iterator["Complex"]: + """Get all complexes from A&AI. + + Call A&AI API to get all complex objects. + + Args: + physical_location_id (str, optional): Unique identifier for physical location, + e.g., CLLI. Defaults to None. + data_center_code (str, optional): Data center code which can be an alternate way + to identify a complex. Defaults to None. + complex_name (str, optional): Gamma complex name for LCP instance. Defaults to None. + identity_url (str, optional): URL of the keystone identity service. Defaults to None. + + Yields: + Complex -- Complex object. Can not yield anything if any complex with given filter + parameters doesn't exist + + """ + filter_parameters: dict = cls.filter_none_key_values( + { + "physical-location-id": physical_location_id, + "data-center-code": data_center_code, + "complex-name": complex_name, + "identity-url": identity_url, + } + ) + url: str = (f"{cls.get_all_url()}?{urlencode(filter_parameters)}") + for complex_json in cls.send_message_json("GET", + "get cloud regions", + url).get("complex", []): + yield cls.create_from_api_response(complex_json) + + @classmethod + def get_by_physical_location_id(cls, physical_location_id: str) -> "Complex": + """Get complex by physical location id. + + Args: + physical_location_id (str): Physical location id of Complex + + Returns: + Complex: Complex object + + Raises: + ResourceNotFound: Complex with given physical location id not found + + """ + response = cls.send_message_json("GET", + "Get complex with physical location id: " + f"{physical_location_id}", + f"{cls.base_url}{cls.api_version}/cloud-infrastructure/" + f"complexes/complex/{physical_location_id}") + return cls.create_from_api_response(response) + + @classmethod + def create_from_api_response(cls, + api_response: Dict[str, Any]) -> "Complex": + """Create complex object using given A&AI API response JSON. + + Args: + api_response (Dict[str, Any]): Complex A&AI API response + + Returns: + Complex: Complex object created from given response + + """ + return cls( + name=api_response.get("complex-name"), + physical_location_id=api_response["physical-location-id"], + data_center_code=api_response.get("data-center-code"), + identity_url=api_response.get("identity-url"), + resource_version=api_response.get("resource-version"), + physical_location_type=api_response.get("physical-location-type"), + street1=api_response.get("street1"), + street2=api_response.get("street2"), + city=api_response.get("city"), + state=api_response.get("state"), + postal_code=api_response.get("postal-code"), + country=api_response.get("country"), + region=api_response.get("region"), + latitude=api_response.get("latitude"), + longitude=api_response.get("longitude"), + elevation=api_response.get("elevation"), + lata=api_response.get("lata"), + timezone=api_response.get("time-zone"), + data_owner=api_response.get("data-owner"), + data_source=api_response.get("data-source"), + data_source_version=api_response.get("data-source-version") + ) + + def delete(self) -> None: + """Delete complex.""" + self.send_message( + "DELETE", + f"Delete {self.physical_location_id} complex", + f"{self.url}?resource-version={self.resource_version}" + ) diff --git a/src/onapsdk/aai/cloud_infrastructure/geo_region.py b/src/onapsdk/aai/cloud_infrastructure/geo_region.py new file mode 100644 index 0000000..32ff820 --- /dev/null +++ b/src/onapsdk/aai/cloud_infrastructure/geo_region.py @@ -0,0 +1,191 @@ +"""Geo region 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, Optional + +from onapsdk.utils.jinja import jinja_env + +from ..aai_element import AaiResource + + +class GeoRegion(AaiResource): # pylint: disable=too-many-instance-attributes + """Geo region class.""" + + def __init__(self, + geo_region_id: str, + *, + geo_region_name: str = "", + geo_region_type: str = "", + geo_region_role: str = "", + geo_region_function: str = "", + data_owner: str = "", + data_source: str = "", + data_source_version: str = "", + resource_version: str = "", + ) -> None: + """Geo region init. + + Args: + geo_region_id (str): UUID, key for geo-region object. + geo_region_name (str, optional): Name of geo-region. Defaults to "". + geo_region_type (str, optional): Type of geo-region. Defaults to "". + geo_region_role (str, optional): Role of geo-region. Defaults to "". + geo_region_function (str, optional): Function of geo-region. Defaults to "". + data_owner (str, optional): Identifies the entity that is responsible managing + this inventory object. Defaults to "". + data_source (str, optional): Identifies the upstream source of the data. + Defaults to "". + data_source_version (str, optional): Identifies the version of + the upstream source. Defaults to "". + resource_version (str, optional): Resource version. Defaults to "". + + """ + super().__init__() + self.geo_region_id: str = geo_region_id + self.geo_region_name: str = geo_region_name + self.geo_region_type: str = geo_region_type + self.geo_region_role: str = geo_region_role + self.geo_region_function: str = geo_region_function + self.data_owner: str = data_owner + self.data_source: str = data_source + self.data_source_version: str = data_source_version + self.resource_version: str = resource_version + + def __repr__(self) -> str: + """Geo region object representation. + + Returns: + str: Human readable string contains most important information about geo region. + + """ + return ( + f"GeoRegion(geo_region_id={self.geo_region_id})" + ) + + @property + def url(self) -> str: + """Geo region's url. + + Returns: + str: Geo Region's url + + """ + return (f"{self.base_url}{self.api_version}/cloud-infrastructure/" + f"geo-regions/geo-region/{self.geo_region_id}") + + @classmethod + def get_all_url(cls, *args, **kwargs) -> str: # pylint: disable=arguments-differ + """Return url to get all geo regions. + + Returns: + str: Url to get all geo regions + + Raises: + ResourceNotFound: No geo regions found + + """ + return f"{cls.base_url}{cls.api_version}/cloud-infrastructure/geo-regions" + + @classmethod + def get_all(cls) -> Iterator["GeoRegion"]: + """Get all geo regions. + + Yields: + GeoRegion: Geo region + + """ + for geo_region_data in cls.send_message_json("GET", + "Get all geo regions", + cls.get_all_url()).get("geo-region", []): + yield cls(geo_region_id=geo_region_data["geo-region-id"], + geo_region_name=geo_region_data.get("geo-region-name", ""), + geo_region_type=geo_region_data.get("geo-region-type", ""), + geo_region_role=geo_region_data.get("geo-region-role", ""), + geo_region_function=geo_region_data.get("geo-region-function", ""), + data_owner=geo_region_data.get("data-owner", ""), + data_source=geo_region_data.get("data-source", ""), + data_source_version=geo_region_data.get("data-source-version", ""), + resource_version=geo_region_data.get("resource-version", "")) + + @classmethod + def get_by_geo_region_id(cls, geo_region_id: str) -> "GeoRegion": + """Get geo region by it's id. + + Args: + geo_region_id (str): Geo region id + + Returns: + GeoRegion: Geo region + + """ + resp = cls.send_message_json("GET", + f"Get geo region with {geo_region_id} id", + f"{cls.get_all_url()}/geo-region/{geo_region_id}") + return GeoRegion(resp["geo-region-id"], + geo_region_name=resp.get("geo-region-name", ""), + geo_region_type=resp.get("geo-region-type", ""), + geo_region_role=resp.get("geo-region-role", ""), + geo_region_function=resp.get("geo-region-function", ""), + data_owner=resp.get("data-owner", ""), + data_source=resp.get("data-source", ""), + data_source_version=resp.get("data-source-version", ""), + resource_version=resp["resource-version"]) + + @classmethod + def create(cls, # pylint: disable=too-many-arguments + geo_region_id: str, + geo_region_name: Optional[str] = None, + geo_region_type: Optional[str] = None, + geo_region_role: Optional[str] = None, + geo_region_function: Optional[str] = None, + data_owner: Optional[str] = None, + data_source: Optional[str] = None, + data_source_version: Optional[str] = None) -> "GeoRegion": + """Create geo region. + + Args: + geo_region_id (str): UUID, key for geo-region object. + geo_region_name (Optional[str], optional): Name of geo-region. Defaults to None. + geo_region_type (Optional[str], optional): Type of geo-region. Defaults to None. + geo_region_role (Optional[str], optional): Role of geo-region. Defaults to None. + geo_region_function (Optional[str], optional): Function of geo-region. + Defaults to None. + data_owner (Optional[str], optional): Identifies the entity that is + responsible managing this inventory object.. Defaults to None. + data_source (Optional[str], optional): Identifies the upstream source of the data. + Defaults to None. + data_source_version (Optional[str], optional): Identifies the version of + the upstream source. Defaults to None. + + Returns: + GeoRegion: Geo region object + + """ + cls.send_message( + "PUT", + "Create geo region", + f"{cls.get_all_url()}/geo-region/{geo_region_id}", + data=jinja_env() + .get_template("geo_region_create.json.j2") + .render(geo_region_id=geo_region_id, + geo_region_name=geo_region_name, + geo_region_type=geo_region_type, + geo_region_role=geo_region_role, + geo_region_function=geo_region_function, + data_owner=data_owner, + data_source=data_source, + data_source_version=data_source_version), + ) + return cls.get_by_geo_region_id(geo_region_id) diff --git a/src/onapsdk/aai/cloud_infrastructure/tenant.py b/src/onapsdk/aai/cloud_infrastructure/tenant.py new file mode 100644 index 0000000..13d9aec --- /dev/null +++ b/src/onapsdk/aai/cloud_infrastructure/tenant.py @@ -0,0 +1,101 @@ +"""A&AI Tenant 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 onapsdk.aai.cloud_infrastructure.cloud_region import CloudRegion +from ..aai_element import AaiResource + + +class Tenant(AaiResource): + """Tenant class.""" + + def __init__(self, # pylint: disable=too-many-arguments + cloud_region: "CloudRegion", + tenant_id: str, + tenant_name: str, + tenant_context: str = None, + resource_version: str = None): + """Tenant object initialization. + + Tenant object represents A&AI Tenant resource. + + Args: + cloud_region (str): Cloud region object + tenant_id (str): Unique Tenant ID + tenant_name (str): Tenant name + tenant_context (str, optional): Tenant context. Defaults to None. + resource_version (str, optional): Tenant resource version. Defaults to None. + + """ + super().__init__() + self.cloud_region: "CloudRegion" = cloud_region + self.tenant_id: str = tenant_id + self.name: str = tenant_name + self.context: str = tenant_context + self.resource_version: str = resource_version + + def __repr__(self) -> str: + """Tenant repr. + + Returns: + str: Human readable Tenant object description + + """ + return ( + f"Tenant(tenant_id={self.tenant_id}, tenant_name={self.name}, " + f"tenant_context={self.context}, " + f"resource_version={self.resource_version}, " + f"cloud_region={self.cloud_region.cloud_region_id})" + ) + + @classmethod + def get_all_url(cls, cloud_region: "CloudRegion") -> str: # pylint: disable=arguments-differ + """Return an url to get all tenants for given cloud region. + + Args: + cloud_region (CloudRegion): Cloud region object + + Returns: + str: Url to get all tenants + + """ + return (f"{cls.base_url}{cls.api_version}/cloud-infrastructure/cloud-regions/cloud-region/" + f"{cloud_region.cloud_owner}/{cloud_region.cloud_region_id}" + f"/tenants/") + + @property + def url(self) -> str: + """Tenant url. + + Returns: + str: Url which can be used to update or delete tenant. + + """ + return ( + f"{self.base_url}{self.api_version}/cloud-infrastructure/cloud-regions/cloud-region/" + f"{self.cloud_region.cloud_owner}/{self.cloud_region.cloud_region_id}" + f"/tenants/tenant/{self.tenant_id}?" + f"resource-version={self.resource_version}" + ) + + def delete(self) -> None: + """Delete tenant. + + Remove tenant from cloud region. + + """ + return self.send_message( + "DELETE", + f"Remove tenant {self.name} from {self.cloud_region.cloud_region_id} cloud region", + url=self.url, + ) |