diff --git a/singlestoredb/management/region.py b/singlestoredb/management/region.py index 3b0c2f30a..546070e4b 100644 --- a/singlestoredb/management/region.py +++ b/singlestoredb/management/region.py @@ -4,6 +4,7 @@ from typing import Optional from .manager import Manager +from .utils import NamedList from .utils import vars_to_str @@ -65,3 +66,94 @@ def from_dict(cls, obj: Dict[str, str], manager: Manager) -> 'Region': ) out._manager = manager return out + + +class RegionManager(Manager): + """ + SingleStoreDB region manager. + + This class should be instantiated using :func:`singlestoredb.manage_regions`. + + Parameters + ---------- + access_token : str, optional + The API key or other access token for the workspace management API + version : str, optional + Version of the API to use + base_url : str, optional + Base URL of the workspace management API + + See Also + -------- + :func:`singlestoredb.manage_regions` + """ + + #: Object type + obj_type = 'region' + + def list_regions(self) -> NamedList[Region]: + """ + List all available regions. + + Returns + ------- + NamedList[Region] + List of available regions + + Raises + ------ + ManagementError + If there is an error getting the regions + """ + res = self._get('regions') + return NamedList( + [Region.from_dict(item, self) for item in res.json()], + ) + + def list_shared_tier_regions(self) -> NamedList[Region]: + """ + List regions that support shared tier workspaces. + + Returns + ------- + NamedList[Region] + List of regions that support shared tier workspaces + + Raises + ------ + ManagementError + If there is an error getting the regions + """ + res = self._get('regions/sharedtier') + return NamedList( + [Region.from_dict(item, self) for item in res.json()], + ) + + +def manage_regions( + access_token: Optional[str] = None, + version: Optional[str] = None, + base_url: Optional[str] = None, +) -> RegionManager: + """ + Retrieve a SingleStoreDB region manager. + + Parameters + ---------- + access_token : str, optional + The API key or other access token for the workspace management API + version : str, optional + Version of the API to use + base_url : str, optional + Base URL of the workspace management API + + Returns + ------- + :class:`RegionManager` + + """ + return RegionManager( + access_token=access_token, + version=version, + base_url=base_url, + ) diff --git a/singlestoredb/tests/test_management.py b/singlestoredb/tests/test_management.py index 222c80001..bbec64825 100755 --- a/singlestoredb/tests/test_management.py +++ b/singlestoredb/tests/test_management.py @@ -13,6 +13,8 @@ import singlestoredb as s2 from singlestoredb.management.job import Status from singlestoredb.management.job import TargetType +from singlestoredb.management.region import Region +from singlestoredb.management.utils import NamedList TEST_DIR = pathlib.Path(os.path.dirname(__file__)) @@ -1372,3 +1374,107 @@ def test_file_object(self): # Cleanup space.remove('obj_test_2.ipynb') + + +@pytest.mark.management +class TestRegions(unittest.TestCase): + """Test cases for region management.""" + + manager = None + + @classmethod + def setUpClass(cls): + """Set up the test environment.""" + cls.manager = s2.manage_regions() + + @classmethod + def tearDownClass(cls): + """Clean up the test environment.""" + cls.manager = None + + def test_list_regions(self): + """Test listing all regions.""" + regions = self.manager.list_regions() + + # Verify we get a NamedList + assert isinstance(regions, NamedList) + + # Verify we have at least one region + assert len(regions) > 0 + + # Verify region properties + region = regions[0] + assert isinstance(region, Region) + assert hasattr(region, 'id') + assert hasattr(region, 'name') + assert hasattr(region, 'provider') + + # Verify provider values + providers = {x.provider for x in regions} + assert 'Azure' in providers or 'GCP' in providers or 'AWS' in providers + + # Verify region can be accessed by name or ID + region_by_name = regions[region.name] + region_by_id = regions[region.id] + assert region_by_name == region_by_id + assert region_by_name.id == region.id + assert region_by_name.name == region.name + assert region_by_name.provider == region.provider + + def test_list_shared_tier_regions(self): + """Test listing shared tier regions.""" + regions = self.manager.list_shared_tier_regions() + + # Verify we get a NamedList + assert isinstance(regions, NamedList) + + # Verify region properties if we have any shared tier regions + if regions: + region = regions[0] + assert isinstance(region, Region) + assert hasattr(region, 'id') + assert hasattr(region, 'name') + assert hasattr(region, 'provider') + + # Verify provider values + providers = {x.provider for x in regions} + assert any(p in providers for p in ['Azure', 'GCP', 'AWS']) + + # Verify region can be accessed by name or ID + region_by_name = regions[region.name] + region_by_id = regions[region.id] + assert region_by_name == region_by_id + assert region_by_name.id == region.id + assert region_by_name.name == region.name + assert region_by_name.provider == region.provider + + def test_str_repr(self): + """Test string representation of regions.""" + regions = self.manager.list_regions() + if not regions: + self.skipTest('No regions available for testing') + + region = regions[0] + + # Test __str__ + s = str(region) + assert region.id in s + assert region.name in s + assert region.provider in s + + # Test __repr__ + assert repr(region) == str(region) + + def test_no_manager(self): + """Test behavior when manager is not available.""" + regions = self.manager.list_regions() + if not regions: + self.skipTest('No regions available for testing') + + region = regions[0] + region._manager = None + + # Verify from_dict class method + with self.assertRaises(s2.ManagementError) as cm: + Region.get_shared_tier_regions(None) + assert 'No workspace manager' in str(cm.exception)