diff options
author | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2022-10-17 12:46:49 +0000 |
---|---|---|
committer | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2022-10-17 14:05:09 +0000 |
commit | f2adf542e878c96895210f97ebf1ebb763b2f465 (patch) | |
tree | 91fc0faeb3436e723d07aed1f38ce59a6e7cc7c5 /src/onapsdk/cps | |
parent | 49071a0d0425ef67fa552dbf14c81e5a11cc49e7 (diff) |
Release ONAP SDKv10.2
Issue-ID: INT-2150
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: I650047c599a5aae6de7c6b42d38e34aea88578e2
Diffstat (limited to 'src/onapsdk/cps')
-rw-r--r-- | src/onapsdk/cps/__init__.py | 18 | ||||
-rw-r--r-- | src/onapsdk/cps/anchor.py | 193 | ||||
-rw-r--r-- | src/onapsdk/cps/cps_element.py | 24 | ||||
-rw-r--r-- | src/onapsdk/cps/dataspace.py | 193 | ||||
-rw-r--r-- | src/onapsdk/cps/schemaset.py | 73 |
5 files changed, 501 insertions, 0 deletions
diff --git a/src/onapsdk/cps/__init__.py b/src/onapsdk/cps/__init__.py new file mode 100644 index 0000000..63d5dd5 --- /dev/null +++ b/src/onapsdk/cps/__init__.py @@ -0,0 +1,18 @@ +"""ONAP SDK CPS 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 .anchor import Anchor # pylint: disable=unused-import +from .dataspace import Dataspace # pylint: disable=unused-import +from .schemaset import SchemaSet, SchemaSetModuleReference # pylint: disable=unused-import diff --git a/src/onapsdk/cps/anchor.py b/src/onapsdk/cps/anchor.py new file mode 100644 index 0000000..f02687d --- /dev/null +++ b/src/onapsdk/cps/anchor.py @@ -0,0 +1,193 @@ +"""ONAP SDK CPS anchor 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, TYPE_CHECKING + +from .cps_element import CpsElement + +if TYPE_CHECKING: + from .schemaset import SchemaSet # pylint: disable=cyclic-import + + +class Anchor(CpsElement): + """CPS anchor class.""" + + def __init__(self, name: str, schema_set: "SchemaSet") -> None: + """Initialise CPS anchor object. + + Args: + name (str): Anchor name + schema_set (SchemaSet): Schema set + + """ + super().__init__() + self.name: str = name + self.schema_set: "SchemaSet" = schema_set + + def __repr__(self) -> str: + """Human readable representation of the object. + + Returns: + str: Human readable string + + """ + return f"Anchor(name={self.name}, "\ + f"schema set={self.schema_set.name}, "\ + f"dataspace={self.schema_set.dataspace.name})" + + @property + def url(self) -> str: + """Anchor url. + + Returns: + str: Anchor url + + """ + return f"{self._url}/cps/api/v1/dataspaces/"\ + f"{self.schema_set.dataspace.name}/anchors/{self.name}" + + def delete(self) -> None: + """Delete anchor.""" + self.send_message( + "DELETE", + f"Delete {self.name} anchor", + self.url, + auth=self.auth + ) + + def create_node(self, node_data: str) -> None: + """Create anchor node. + + Fill CPS anchor with a data. + + Args: + node_data (str): Node data. Should be JSON formatted. + + """ + self.send_message( + "POST", + f"Create {self.name} anchor node", + f"{self.url}/nodes", + data=node_data, + auth=self.auth + ) + + def get_node(self, xpath: str, include_descendants: bool = False) -> Dict[Any, Any]: + """Get anchor node data. + + Using XPATH get anchor's node data. + + Args: + xpath (str): Anchor node xpath. + include_descendants (bool, optional): Determies if descendants should be included in + response. Defaults to False. + + Returns: + Dict[Any, Any]: Anchor node data. + + """ + return self.send_message_json( + "GET", + f"Get {self.name} anchor node with {xpath} xpath", + f"{self.url}/node?xpath={xpath}&include-descendants={include_descendants}", + auth=self.auth + ) + + def update_node(self, xpath: str, node_data: str) -> None: + """Update anchor node data. + + Using XPATH update anchor's node data. + + Args: + xpath (str): Anchor node xpath. + node_data (str): Node data. + + """ + self.send_message( + "PATCH", + f"Update {self.name} anchor node with {xpath} xpath", + f"{self.url}/nodes?xpath={xpath}", + data=node_data, + auth=self.auth + ) + + def replace_node(self, xpath: str, node_data: str) -> None: + """Replace anchor node data. + + Using XPATH replace anchor's node data. + + Args: + xpath (str): Anchor node xpath. + node_data (str): Node data. + + """ + self.send_message( + "PUT", + f"Replace {self.name} anchor node with {xpath} xpath", + f"{self.url}/nodes?xpath={xpath}", + data=node_data, + auth=self.auth + ) + + def add_list_node(self, xpath: str, node_data: str) -> None: + """Add an element to the list node of an anchor. + + Args: + xpath (str): Xpath to the list node. + node_data (str): Data to be added. + + """ + self.send_message( + "POST", + f"Add element to {self.name} anchor node with {xpath} xpath", + f"{self.url}/list-nodes?xpath={xpath}", + data=node_data, + auth=self.auth + ) + + def query_node(self, query: str, include_descendants: bool = False) -> Dict[Any, Any]: + """Query CPS anchor data. + + Args: + query (str): Query + include_descendants (bool, optional): Determies if descendants should be included in + response. Defaults to False. + + Returns: + Dict[Any, Any]: Query return values. + + """ + return self.send_message_json( + "GET", + f"Get {self.name} anchor node with {query} query", + f"{self.url}/nodes/query?cps-path={query}&include-descendants={include_descendants}", + auth=self.auth + ) + + def delete_nodes(self, xpath: str) -> None: + """Delete nodes. + + Use XPATH to delete Anchor nodes. + + Args: + xpath (str): Nodes to delete + + """ + self.send_message( + "DELETE", + f"Delete {self.name} anchor nodes with {xpath} xpath", + f"{self.url}/nodes?xpath={xpath}", + auth=self.auth + ) diff --git a/src/onapsdk/cps/cps_element.py b/src/onapsdk/cps/cps_element.py new file mode 100644 index 0000000..27f1faa --- /dev/null +++ b/src/onapsdk/cps/cps_element.py @@ -0,0 +1,24 @@ +"""ONAP SDK CPS element 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.configuration import settings +from onapsdk.onap_service import OnapService + + +class CpsElement(OnapService): + """Mother Class of all CPS elements.""" + + _url: str = settings.CPS_URL + auth: tuple = settings.CPS_AUTH diff --git a/src/onapsdk/cps/dataspace.py b/src/onapsdk/cps/dataspace.py new file mode 100644 index 0000000..e6340d7 --- /dev/null +++ b/src/onapsdk/cps/dataspace.py @@ -0,0 +1,193 @@ +"""ONAP SDK CPS dataspace 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, Iterable + +from .anchor import Anchor +from .cps_element import CpsElement +from .schemaset import SchemaSet, SchemaSetModuleReference + + +class Dataspace(CpsElement): + """CPS dataspace class.""" + + def __init__(self, name: str) -> None: + """Initialize dataspace object. + + Args: + name (str): Dataspace name + + """ + super().__init__() + self.name: str = name + + def __repr__(self) -> str: + """Human readable representation of the object. + + Returns: + str: Human readable string + + """ + return f"Dataspace(name={self.name})" + + @property + def url(self) -> str: + """Dataspace url. + + Returns: + str: Dataspace url + + """ + return f"{self._url}/cps/api/v1/dataspaces/{self.name}" + + @classmethod + def create(cls, dataspace_name: str) -> "Dataspace": + """Create dataspace with given name. + + Args: + dataspace_name (str): Dataspace name + + Returns: + Dataspace: Newly created dataspace + + """ + cls.send_message( + "POST", + f"Create {dataspace_name} dataspace", + f"{cls._url}/cps/api/v1/dataspaces?dataspace-name={dataspace_name}", + auth=cls.auth + ) + return Dataspace(dataspace_name) + + def create_anchor(self, schema_set: SchemaSet, anchor_name: str) -> Anchor: + """Create anchor. + + Args: + schema_set (SchemaSet): Schema set object which is going to be used to create anchor. + anchor_name (str): Anchor name + + Returns: + Anchor: Created anchor + + """ + self.send_message( + "POST", + "Get all CPS dataspace schemasets", + f"{self.url}/anchors/?schema-set-name={schema_set.name}&anchor-name={anchor_name}", + auth=self.auth + ) + return Anchor(name=anchor_name, schema_set=schema_set) + + def get_anchors(self) -> Iterable[Anchor]: + """Get all dataspace's anchors. + + Iterable of related with dataspace anchors. + + Yields: + Iterator[Anchor]: Anchor object + + """ + for anchor_data in self.send_message_json(\ + "GET",\ + "Get all CPS dataspace anchors",\ + f"{self.url}/anchors",\ + auth=self.auth\ + ): + yield Anchor(name=anchor_data["name"], + schema_set=SchemaSet(name=anchor_data["schemaSetName"], + dataspace=self)) + + def get_anchor(self, anchor_name: str) -> Anchor: + """Get dataspace anchor by name. + + To get anchor there is no need to use `SchemaSet` object, but to create anchor it it. + + Args: + anchor_name (str): Anchor name. + + Returns: + Anchor: Anchor object + + """ + anchor_data: Dict[str, Any] = self.send_message_json( + "GET", + f"Get {anchor_name} anchor", + f"{self.url}/anchors/{anchor_name}", + auth=self.auth + ) + return Anchor(name=anchor_data["name"], + schema_set=SchemaSet(name=anchor_data["schemaSetName"], + dataspace=self)) + + def get_schema_set(self, schema_set_name: str) -> SchemaSet: + """Get schema set by name. + + Args: + schema_set_name (str): Schema set name + + Returns: + SchemaSet: Schema set object + + """ + schema_set_data: Dict[str, Any] = self.send_message_json( + "GET", + "Get all CPS dataspace schemasets", + f"{self._url}/cps/api/v1/dataspaces/{self.name}/schema-sets/{schema_set_name}", + auth=self.auth + ) + return SchemaSet( + name=schema_set_data["name"], + dataspace=self, + module_references=[ + SchemaSetModuleReference( + name=module_reference_data["name"], + namespace=module_reference_data["namespace"], + revision=module_reference_data["revision"] + ) for module_reference_data in schema_set_data["moduleReferences"] + ] + ) + + def create_schema_set(self, schema_set_name: str, schema_set: bytes) -> SchemaSet: + """Create schema set. + + Create CPS schema set in dataspace + + Args: + schema_set_name (str): Schema set name + schema_set (bytes): Schema set YANG + + Returns: + SchemaSet: Created schema set object + + """ + self.send_message( + "POST", + "Create schema set", + f"{self._url}/cps/api/v1/dataspaces/{self.name}/schema-sets/", + files={"file": schema_set}, + data={"schema-set-name": schema_set_name}, + headers={}, # Leave headers empty to fill it correctly by `requests` library + auth=self.auth + ) + return self.get_schema_set(schema_set_name) + + def delete(self) -> None: + """Delete dataspace.""" + self.send_message( + "DELETE", + f"Delete {self.name} dataspace", + f"{self._url}/cps/api/v1/dataspaces/{self.name}", + auth=self.auth + ) diff --git a/src/onapsdk/cps/schemaset.py b/src/onapsdk/cps/schemaset.py new file mode 100644 index 0000000..650d12d --- /dev/null +++ b/src/onapsdk/cps/schemaset.py @@ -0,0 +1,73 @@ +"""ONAP SDK CPS schemaset 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 List, Optional, TYPE_CHECKING + +from .cps_element import CpsElement + +if TYPE_CHECKING: + from .dataspace import Dataspace # pylint: disable=cyclic-import + + +@dataclass +class SchemaSetModuleReference: + """Schema set module reference dataclass. + + Stores all information about module reference. + """ + + name: str + namespace: str + revision: str + + +class SchemaSet(CpsElement): + """Schema set class.""" + + def __init__(self, + name: str, + dataspace: "Dataspace", + module_references: Optional[List[SchemaSetModuleReference]] = None) -> None: + """Initialize schema set class object. + + Args: + name (str): Schema set name + dataspace (Dataspace): Dataspace on which schema set was created. + module_references (Optional[List[SchemaSetModuleReference]], optional): + List of module references. Defaults to None. + """ + super().__init__() + self.name: str = name + self.dataspace: "Dataspace" = dataspace + self.module_refences: List[SchemaSetModuleReference] = module_references \ + if module_references else [] + + def __repr__(self) -> str: + """Human readable representation of the object. + + Returns: + str: Human readable string + + """ + return f"SchemaSet(name={self.name}, dataspace={self.dataspace.name})" + + def delete(self) -> None: + """Delete schema set.""" + self.send_message( + "DELETE", + f"Delete {self.name} schema set", + f"{self._url}/cps/api/v1/dataspaces/{self.dataspace.name}/schema-sets/{self.name}" + ) |