refactor(resources): one canonical layout, scoped bases, dedup schemas

examples/dashboard/app.py

@@ -35,7 +35,7 @@
     WorkspaceCreate,
     WorkspaceUpdate,
 )
-from codesphere.resources.workspace.envVars import EnvVar  # noqa: E402
+from codesphere.resources.workspace.env_vars import EnvVar  # noqa: E402
 from codesphere.resources.workspace.landscape import (  # noqa: E402
     PipelineStage,
     ProfileBuilder,

src/codesphere/__init__.py

@@ -51,7 +51,7 @@
     WorkspaceStatus,
     WorkspaceUpdate,
 )
-from .resources.workspace.envVars import EnvVar
+from .resources.workspace.env_vars import EnvVar
 
 logging.getLogger("codesphere").addHandler(logging.NullHandler())
 

src/codesphere/core/__init__.py

@@ -1,4 +1,11 @@
-from .base import BoundModel, CamelModel, ResourceBase, ResourceList
+from .base import (
+    BoundModel,
+    CamelModel,
+    ResourceBase,
+    ResourceList,
+    TeamScopedResource,
+    WorkspaceScopedResource,
+)
 from .handler import execute_operation
 from .operations import APIOperation, StreamOperation
 
@@ -9,5 +16,7 @@
     "ResourceBase",
     "ResourceList",
     "StreamOperation",
+    "TeamScopedResource",
+    "WorkspaceScopedResource",
     "execute_operation",
 ]

src/codesphere/core/base.py

@@ -32,6 +32,22 @@ async def _execute(
         )
 
 
+class TeamScopedResource(ResourceBase):
+    """Base for managers operating within one team."""
+
+    def __init__(self, http_client: APIHttpClient, team_id: int):
+        super().__init__(http_client)
+        self.team_id = team_id
+
+
+class WorkspaceScopedResource(ResourceBase):
+    """Base for managers operating within one workspace."""
+
+    def __init__(self, http_client: APIHttpClient, workspace_id: int):
+        super().__init__(http_client)
+        self.workspace_id = workspace_id
+
+
 class CamelModel(BaseModel):
     model_config = ConfigDict(
         alias_generator=to_camel,

src/codesphere/resources/team/__init__.py

@@ -1,9 +1,10 @@
-from .domain.resources import (
+from .domain import (
     CustomDomainConfig,
     Domain,
     DomainBase,
     DomainRouting,
     DomainVerificationStatus,
+    TeamDomainManager,
 )
 from .resources import TeamsResource
 from .schemas import Team, TeamBase, TeamCreate
@@ -30,6 +31,7 @@
     "Team",
     "TeamBase",
     "TeamCreate",
+    "TeamDomainManager",
     "TeamUsageManager",
     "TeamsResource",
     "UsageEventsResponse",

src/codesphere/resources/team/domain/__init__.py

@@ -0,0 +1,19 @@
+from .resources import TeamDomainManager
+from .schemas import (
+    CustomDomainConfig,
+    Domain,
+    DomainBase,
+    DomainRouting,
+    DomainVerificationStatus,
+    RoutingMap,
+)
+
+__all__ = [
+    "CustomDomainConfig",
+    "Domain",
+    "DomainBase",
+    "DomainRouting",
+    "DomainVerificationStatus",
+    "RoutingMap",
+    "TeamDomainManager",
+]

src/codesphere/resources/team/domain/operations.py

@@ -2,8 +2,7 @@
 
 from ....core.base import ResourceList
 from ....core.operations import APIOperation
-from .resources import Domain
-from .schemas import CustomDomainConfig, DomainVerificationStatus
+from .schemas import CustomDomainConfig, Domain, DomainVerificationStatus
 
 _LIST_OP = APIOperation(
     method="GET",

src/codesphere/resources/team/domain/resources.py

@@ -1,52 +1,30 @@
-from __future__ import annotations
+from ....core.base import TeamScopedResource
+from .operations import _CREATE_OP, _GET_OP, _LIST_OP, _UPDATE_OP, _UPDATE_WS_OP
+from .schemas import CustomDomainConfig, Domain, DomainRouting, RoutingMap
 
-import logging
 
-from ....core.base import BoundModel
-from ....utils import update_model_fields
-from .schemas import (
-    CustomDomainConfig,
-    DomainBase,
-    DomainRouting,
-    DomainVerificationStatus,
-    RoutingMap,
-)
+class TeamDomainManager(TeamScopedResource):
+    async def list(self) -> list[Domain]:
+        result = await self._execute(_LIST_OP, team_id=self.team_id)
+        return result.root
 
-log = logging.getLogger(__name__)
+    async def get(self, name: str) -> Domain:
+        return await self._execute(_GET_OP, team_id=self.team_id, name=name)
 
+    async def create(self, name: str) -> Domain:
+        return await self._execute(_CREATE_OP, team_id=self.team_id, name=name)
 
-class Domain(DomainBase, BoundModel):
-    async def update(self, data: CustomDomainConfig) -> Domain:
-        from .operations import _UPDATE_OP
-
-        response = await self._execute(
-            _UPDATE_OP, team_id=self.team_id, name=self.name, data=data
+    async def update(self, name: str, config: CustomDomainConfig) -> Domain:
+        return await self._execute(
+            _UPDATE_OP, team_id=self.team_id, name=name, data=config
         )
-        update_model_fields(target=self, source=response)
-        return response
 
     async def update_workspace_connections(
-        self, connections: DomainRouting | RoutingMap
+        self, name: str, connections: DomainRouting | RoutingMap
     ) -> Domain:
-        from .operations import _UPDATE_WS_OP
-
         payload = (
             connections.root if isinstance(connections, DomainRouting) else connections
         )
-        response = await self._execute(
-            _UPDATE_WS_OP, team_id=self.team_id, name=self.name, data=payload
+        return await self._execute(
+            _UPDATE_WS_OP, team_id=self.team_id, name=name, data=payload
         )
-        update_model_fields(target=self, source=response)
-        return response
-
-    async def verify_status(self) -> DomainVerificationStatus:
-        from .operations import _VERIFY_OP
-
-        response = await self._execute(_VERIFY_OP, team_id=self.team_id, name=self.name)
-        update_model_fields(target=self.domain_verification_status, source=response)
-        return response
-
-    async def delete(self) -> None:
-        from .operations import _DELETE_OP
-
-        await self._execute(_DELETE_OP, team_id=self.team_id, name=self.name)

src/codesphere/resources/team/domain/schemas.py

@@ -2,7 +2,8 @@
 
 from pydantic import Field, RootModel
 
-from ....core.base import CamelModel
+from ....core.base import BoundModel, CamelModel
+from ....utils import update_model_fields
 
 type RoutingMap = dict[str, list[int]]
 
@@ -48,3 +49,40 @@ class DomainBase(CamelModel):
     domain_verification_status: DomainVerificationStatus
     custom_config_revision: int | None = None
     custom_config: CustomDomainConfig | None = None
+
+
+class Domain(DomainBase, BoundModel):
+    async def update(self, data: CustomDomainConfig) -> Domain:
+        from .operations import _UPDATE_OP
+
+        response = await self._execute(
+            _UPDATE_OP, team_id=self.team_id, name=self.name, data=data
+        )
+        update_model_fields(target=self, source=response)
+        return response
+
+    async def update_workspace_connections(
+        self, connections: DomainRouting | RoutingMap
+    ) -> Domain:
+        from .operations import _UPDATE_WS_OP
+
+        payload = (
+            connections.root if isinstance(connections, DomainRouting) else connections
+        )
+        response = await self._execute(
+            _UPDATE_WS_OP, team_id=self.team_id, name=self.name, data=payload
+        )
+        update_model_fields(target=self, source=response)
+        return response
+
+    async def verify_status(self) -> DomainVerificationStatus:
+        from .operations import _VERIFY_OP
+
+        response = await self._execute(_VERIFY_OP, team_id=self.team_id, name=self.name)
+        update_model_fields(target=self.domain_verification_status, source=response)
+        return response
+
+    async def delete(self) -> None:
+        from .operations import _DELETE_OP
+
+        await self._execute(_DELETE_OP, team_id=self.team_id, name=self.name)

src/codesphere/resources/team/schemas.py

@@ -3,8 +3,8 @@
 from functools import cached_property
 
 from ...core.base import BoundModel, CamelModel
-from .domain.manager import TeamDomainManager
-from .usage.manager import TeamUsageManager
+from .domain.resources import TeamDomainManager
+from .usage.resources import TeamUsageManager
 
 
 class TeamCreate(CamelModel):

src/codesphere/resources/team/usage/__init__.py

@@ -1,6 +1,6 @@
 """Team usage history resources."""
 
-from .manager import TeamUsageManager
+from .resources import TeamUsageManager
 from .schemas import (
     LandscapeServiceEvent,
     LandscapeServiceSummary,

src/codesphere/resources/team/usage/manager.py → src/codesphere/resources/team/usage/resources.py

@@ -4,8 +4,7 @@
 from datetime import datetime
 from functools import partial
 
-from ....core.base import ResourceBase
-from ....http_client import APIHttpClient
+from ....core.base import TeamScopedResource
 from .operations import _GET_LANDSCAPE_EVENTS_OP, _GET_LANDSCAPE_SUMMARY_OP
 from .schemas import (
     LandscapeServiceEvent,
@@ -15,7 +14,12 @@
 )
 
 
-class TeamUsageManager(ResourceBase):
+def _clamp_page_size(value: int) -> int:
+    """The usage API accepts page sizes between 1 and 100."""
+    return min(max(1, value), 100)
+
+
+class TeamUsageManager(TeamScopedResource):
     """Manager for team usage history operations.
 
     Provides access to landscape service usage summaries and events with
@@ -44,10 +48,6 @@ class TeamUsageManager(ResourceBase):
         ```
     """
 
-    def __init__(self, http_client: APIHttpClient, team_id: int):
-        super().__init__(http_client)
-        self.team_id = team_id
-
     async def get_landscape_summary(
         self,
         begin_date: datetime | str,
@@ -62,7 +62,7 @@ async def get_landscape_summary(
             "endDate": end_date.isoformat()
             if isinstance(end_date, datetime)
             else end_date,
-            "limit": min(max(1, limit), 100),
+            "limit": _clamp_page_size(limit),
             "offset": max(0, offset),
         }
         result = await self._execute(
@@ -83,7 +83,7 @@ async def iter_all_landscape_summary(
         page_size: int = 100,
     ) -> AsyncIterator[LandscapeServiceSummary]:
         offset = 0
-        page_size = min(max(1, page_size), 100)
+        page_size = _clamp_page_size(page_size)
 
         while True:
             response = await self.get_landscape_summary(
@@ -116,7 +116,7 @@ async def get_landscape_events(
             "endDate": end_date.isoformat()
             if isinstance(end_date, datetime)
             else end_date,
-            "limit": min(max(1, limit), 100),
+            "limit": _clamp_page_size(limit),
             "offset": max(0, offset),
         }
         result = await self._execute(
@@ -145,7 +145,7 @@ async def iter_all_landscape_events(
         page_size: int = 100,
     ) -> AsyncIterator[LandscapeServiceEvent]:
         offset = 0
-        page_size = min(max(1, page_size), 100)
+        page_size = _clamp_page_size(page_size)
 
         while True:
             response = await self.get_landscape_events(