KAB-46 add configuration metadata create, list and delete

Author: tomasfejfar

kbcstorage/configurations.py

@@ -5,6 +5,7 @@
 """
 import json
 from kbcstorage.base import Endpoint
+from kbcstorage.configurations_metadata import ConfigurationsMetadata
 
 
 class Configurations(Endpoint):
@@ -22,6 +23,7 @@ def __init__(self, root_url, token, branch_id):
             branch_id (str): The ID of branch to use, use 'default' to work without branch (in main).
         """
         super().__init__(root_url, f"branch/{branch_id}/components", token)
+        self.metadata = ConfigurationsMetadata(root_url, token, branch_id)
 
     def detail(self, component_id, configuration_id):
         """

kbcstorage/configurations_metadata.py

@@ -0,0 +1,115 @@
+"""
+Manages calls to the Storage API relating to configurations
+
+Full documentation https://keboola.docs.apiary.io/#reference/components-and-configurations
+"""
+import json
+from kbcstorage.base import Endpoint
+
+
+class ConfigurationsMetadata(Endpoint):
+    """
+    Configurations Endpoint
+    """
+
+    def __init__(self, root_url, token, branch_id):
+        """
+        Create a Component endpoint.
+
+        Args:
+            root_url (:obj:`str`): The base url for the API.
+            token (:obj:`str`): A storage API key.
+            branch_id (str): The ID of branch to use, use 'default' to work without branch (in main).
+        """
+        super().__init__(root_url, f"branch/{branch_id}/components", token)
+
+    def detail(self, component_id, configuration_id):
+        """
+        Retrieves information about a given configuration.
+
+        Args:
+            component_id (str): The id of the component.
+            configuration_id (str): The id of the configuration.
+
+        Returns:
+            response_body: The parsed json from the HTTP response.
+
+        Raises:
+            requests.HTTPError: If the API request fails.
+        """
+        if not isinstance(component_id, str) or component_id == '':
+            raise ValueError("Invalid component_id '{}'.".format(component_id))
+        if not isinstance(configuration_id, str) or configuration_id == '':
+            raise ValueError("Invalid component_id '{}'.".format(configuration_id))
+        url = '{}/{}/configs/{}'.format(self.base_url, component_id, configuration_id)
+        return self._get(url)
+
+    def delete(self, component_id, configuration_id, metadata_id):
+        """
+        Deletes the configuration.
+
+        Args:
+            component_id (str): The id of the component.
+            configuration_id (str): The id of the configuration.
+
+        Raises:
+            requests.HTTPError: If the API request fails.
+        """
+        if not isinstance(component_id, str) or component_id == '':
+            raise ValueError("Invalid component_id '{}'.".format(component_id))
+        if not isinstance(configuration_id, str) or configuration_id == '':
+            raise ValueError("Invalid component_id '{}'.".format(configuration_id))
+        url = '{}/{}/configs/{}/metadata/{}'.format(self.base_url, component_id, configuration_id, metadata_id)
+        self._delete(url)
+
+    def list(self, component_id, configuration_id):
+        """
+        Lists configurations of the given component.
+
+        Args:
+            component_id (str): The id of the component.
+
+        Raises:
+            requests.HTTPError: If the API request fails.
+        """
+        if not isinstance(component_id, str) or component_id == '':
+            raise ValueError("Invalid component_id '{}'.".format(component_id))
+        url = '{}/{}/configs/{}/metadata'.format(self.base_url, component_id, configuration_id)
+        return self._get(url)
+
+    def create(self, component_id, configuration_id, provider, metadata):
+        """
+        Create a new configuration.
+
+        Args:
+            component_id (str): The id of the component.
+            configuration (str): The id of the configuration.
+            provider (str): The provider of the configuration (currently ignored and "user" is sent).
+            key (str): The key of the configuration.
+            value (str): The value of the configuration.
+        Returns:
+            response_body: The parsed json from the HTTP response.
+
+        Raises:
+            requests.HTTPError: If the API request fails.
+        """
+        if not isinstance(component_id, str) or component_id == '':
+            raise ValueError("Invalid component_id '{}'.".format(component_id))
+        if not isinstance(configuration_id, str) or configuration_id == '':
+            raise ValueError("Invalid component_id '{}'.".format(configuration_id))
+        url = '{}/{}/configs/{}/metadata'.format(self.base_url, component_id, configuration_id)
+        if not isinstance(metadata, list):
+            raise ValueError("Metadata must be a list '{}'.".format(metadata))
+        for metadataItem in metadata:
+            if not isinstance(metadataItem, dict):
+                raise ValueError("Metadata item must be a dictionary '{}'.".format(metadataItem))
+
+        headers = {
+              'Content-Type': 'application/json',
+              'X-StorageApi-Token': self.token
+        }
+        data = {
+            # 'provider': provider, # not yet implemented
+            'metadata': metadata
+        }
+        return self._post(url, data=json.dumps(data), headers=headers)

tests/functional/test_configurations.py

@@ -50,3 +50,56 @@ def testListConfigurations(self):
         with self.subTest():
             with self.assertRaises(exceptions.HTTPError):
                 configurations = self.configurations.list('non-existent-component')
+
+    def testConfigurationMetadata(self):
+        self.configurations.create(
+            component_id=self.TEST_COMPONENT_NAME,
+            configuration_id='test_configuration_metadata',
+            name='test_configuration_metadata',
+        )
+        metadataPayload = [
+            {
+                'key': 'testConfigurationMetadata',
+                'value': 'success',
+            }
+        ]
+        metadataList = self.configurations.metadata.create(
+            component_id=self.TEST_COMPONENT_NAME,
+            configuration_id='test_configuration_metadata',
+            provider='test',
+            metadata=metadataPayload,
+        )
+
+        with (self.subTest('assert metadata create response')):
+            self.assertEqual(1, len(metadataList))
+            metadataItem = metadataList[0]
+            self.assertTrue('id' in metadataItem)
+            # self.assertTrue('provider' in metadata) not yet
+            self.assertTrue('key' in metadataItem)
+            self.assertTrue('value' in metadataItem)
+
+            metadataList = self.configurations.metadata.list(
+                component_id=self.TEST_COMPONENT_NAME,
+                configuration_id='test_configuration_metadata'
+            )
+
+        with (self.subTest('assert metadata list response')):
+            self.assertTrue(len(metadataList) > 0)
+            for metadataList in metadataList:
+                self.assertTrue('id' in metadataList)
+                # self.assertTrue('provider' in metadata) not yet
+                self.assertTrue('key' in metadataList)
+                self.assertTrue('value' in metadataList)
+
+        self.configurations.metadata.delete(
+            component_id=self.TEST_COMPONENT_NAME,
+            configuration_id='test_configuration_metadata',
+            metadata_id=metadataList['id']
+        )
+        metadataList = self.configurations.metadata.list(
+            component_id=self.TEST_COMPONENT_NAME,
+            configuration_id='test_configuration_metadata'
+        )
+
+        with (self.subTest('assert metadata delete means metadata no longer in list')):
+            self.assertTrue(len(metadataList) == 0)