Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 92 additions & 0 deletions singlestoredb/management/region.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from typing import Optional

from .manager import Manager
from .utils import NamedList
from .utils import vars_to_str


Expand Down Expand Up @@ -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,
)
106 changes: 106 additions & 0 deletions singlestoredb/tests/test_management.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__))
Expand Down Expand Up @@ -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)