Merge branch 'main' of github.com:singlestore-labs/singlestore-python

Author: kesmit13

singlestoredb/management/organization.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python
 """SingleStoreDB Cloud Organization."""
+import datetime
 from typing import Dict
 from typing import List
 from typing import Optional
 from typing import Union
 
+from ..exceptions import ManagementError
 from .manager import Manager
 from .utils import vars_to_str
 
@@ -21,6 +23,91 @@ def stringify(x: Union[str, List[str]]) -> str:
     return x
 
 
+class Secret(object):
+    """
+    SingleStoreDB secrets definition.
+
+    This object is not directly instantiated. It is used in results
+    of API calls on the :class:`Organization`. See :meth:`Organization.get_secret`.
+    """
+
+    def __init__(
+        self,
+        id: str,
+        name: str,
+        value: str,
+        created_by: str,
+        created_at: Union[str, datetime.datetime],
+        last_updated_by: str,
+        last_updated_at: Union[str, datetime.datetime],
+        deleted_by: Optional[str] = None,
+        deleted_at: Optional[Union[str, datetime.datetime]] = None,
+    ):
+        # UUID of the secret
+        self.id = id
+
+        # Name of the secret
+        self.name = name
+
+        # Value of the secret
+        self.value = value
+
+        # User who created the secret
+        self.created_by = created_by
+
+        # Time when the secret was created
+        self.created_at = created_at
+
+        # UUID of the user who last updated the secret
+        self.last_updated_by = last_updated_by
+
+        # Time when the secret was last updated
+        self.last_updated_at = last_updated_at
+
+        # UUID of the user who deleted the secret
+        self.deleted_by = deleted_by
+
+        # Time when the secret was deleted
+        self.deleted_at = deleted_at
+
+    @classmethod
+    def from_dict(cls, obj: Dict[str, str]) -> 'Secret':
+        """
+        Construct a Secret from a dictionary of values.
+
+        Parameters
+        ----------
+        obj : dict
+            Dictionary of values
+
+        Returns
+        -------
+        :class:`Secret`
+
+        """
+        out = cls(
+            id=obj['secretID'],
+            name=obj['name'],
+            value=obj['value'],
+            created_by=obj['createdBy'],
+            created_at=obj['createdAt'],
+            last_updated_by=obj['lastUpdatedBy'],
+            last_updated_at=obj['lastUpdatedAt'],
+            deleted_by=obj.get('deletedBy'),
+            deleted_at=obj.get('deletedAt'),
+        )
+
+        return out
+
+    def __str__(self) -> str:
+        """Return string representation."""
+        return vars_to_str(self)
+
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return str(self)
+
+
 class Organization(object):
     """
     Organization in SingleStoreDB Cloud portal.
@@ -55,6 +142,22 @@ def __repr__(self) -> str:
         """Return string representation."""
         return str(self)
 
+    def get_secret(self, name: str) -> Secret:
+        if self._manager is None:
+            raise ManagementError(msg='Organization not initialized')
+
+        res = self._manager._get('secrets', params=dict(name=name))
+
+        secrets = [Secret.from_dict(item) for item in res.json()['secrets']]
+
+        if len(secrets) == 0:
+            raise ManagementError(msg=f'Secret {name} not found')
+
+        if len(secrets) > 1:
+            raise ManagementError(msg=f'Multiple secrets found for {name}')
+
+        return secrets[0]
+
     @classmethod
     def from_dict(
         cls,

singlestoredb/tests/test_management.py

@@ -793,3 +793,60 @@ def test_stage_object(self):
         assert not st.exists('obj_test.sql')
         assert st.exists('obj_test_2.sql')
         assert f1.abspath() == 'obj_test_2.sql'
+
+
+@pytest.mark.management
+class TestSecrets(unittest.TestCase):
+
+    manager = None
+    wg = None
+    password = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.manager = s2.manage_workspaces()
+
+        us_regions = [x for x in cls.manager.regions if 'US' in x.name]
+        cls.password = secrets.token_urlsafe(20)
+
+        name = clean_name(secrets.token_urlsafe(20)[:20])
+
+        cls.wg = cls.manager.create_workspace_group(
+            f'wg-test-{name}',
+            region=random.choice(us_regions).id,
+            admin_password=cls.password,
+            firewall_ranges=['0.0.0.0/0'],
+        )
+
+    @classmethod
+    def tearDownClass(cls):
+        if cls.wg is not None:
+            cls.wg.terminate(force=True)
+        cls.wg = None
+        cls.manager = None
+        cls.password = None
+
+    def test_get_secret(self):
+        # manually create secret and then get secret
+        # try to delete the secret if it exists
+        try:
+            secret = self.manager.organizations.current.get_secret('secret_name')
+
+            secret_id = secret.id
+
+            self.manager._delete(f'secrets/{secret_id}')
+        except s2.ManagementError:
+            pass
+
+        self.manager._post(
+            'secrets',
+            json=dict(
+                name='secret_name',
+                value='secret_value',
+            ),
+        )
+
+        secret = self.manager.organizations.current.get_secret('secret_name')
+
+        assert secret.name == 'secret_name'
+        assert secret.value == 'secret_value'