Feature/workspaces

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v0.2.0 (2025-07-16)
+
+### Feat
+
+- **codesphere/workspace**: add workspace resources
+
 ## v0.1.1 (2025-07-16)
 
 ### Fix

Makefile

@@ -27,7 +27,7 @@ commit: ## Starts Commitizen for a guided commit message
 
 lint: ## Checks code quality with ruff
 	@echo ">>> Checking code quality with ruff..."
-	uv run ruff check src
+	uv run ruff check src --fix
 
 format: ## Formats code with ruff
 	@echo ">>> Formatting code with ruff..."

examples/workspaces/create_workspace.py

@@ -0,0 +1,28 @@
+import asyncio
+import pprint
+from codesphere import CodesphereSDK, WorkspaceCreate
+
+
+async def main():
+    """Creates a new workspace in a specific team."""
+    team_id = 12345
+
+    async with CodesphereSDK() as sdk:
+        print(f"--- Creating a new workspace in team {team_id} ---")
+
+        workspace_data = WorkspaceCreate(
+            name="my-new-sdk-workspace-3",
+            planId=8,
+            teamId=int(team_id),
+            isPrivateRepo=True,
+            replicas=1,
+        )
+
+        created_workspace = await sdk.workspaces.create(data=workspace_data)
+
+        print("\n--- Details of successfully created workspace ---")
+        pprint.pprint(created_workspace.model_dump())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/workspaces/delete_workspace.py

@@ -0,0 +1,25 @@
+import asyncio
+from codesphere import CodesphereSDK
+
+
+async def main():
+    """Deletes a specific workspace."""
+
+    workspace_id_to_delete = 12345
+
+    async with CodesphereSDK() as sdk:
+        print(f"--- Fetching workspace with ID: {workspace_id_to_delete} ---")
+        workspace_to_delete = await sdk.workspaces.get(
+            workspace_id=workspace_id_to_delete
+        )
+
+        print(f"\n--- Deleting workspace: '{workspace_to_delete.name}' ---")
+
+        # This is a destructive action!
+        await workspace_to_delete.delete()
+
+        print(f"Workspace '{workspace_to_delete.name}' has been successfully deleted.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/workspaces/list_workspaces.py

@@ -0,0 +1,20 @@
+import asyncio
+import pprint
+from codesphere import CodesphereSDK
+
+
+async def main():
+    """Fetches a team and lists all workspaces within it."""
+    async with CodesphereSDK() as sdk:
+        teams = await sdk.teams.list()
+        workspaces = await sdk.workspaces.list_by_team(team_id=teams[0].id)
+
+        for workspace in workspaces:
+            pprint.pprint(workspace.model_dump())
+        print(f"Found {len(workspaces)} workspace(s):")
+        for ws in workspaces:
+            print(f"  - ID: {ws.id}, Name: {ws.name}, Status: {await ws.get_status()}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/workspaces/update_workspace.py

@@ -0,0 +1,28 @@
+import asyncio
+import pprint
+from codesphere import CodesphereSDK, WorkspaceUpdate
+
+
+async def main():
+    """Fetches a workspace and updates its name."""
+    workspace_id_to_update = 12245
+
+    async with CodesphereSDK() as sdk:
+        print(f"--- Fetching workspace with ID: {workspace_id_to_update} ---")
+        workspace = await sdk.workspaces.get(workspace_id=workspace_id_to_update)
+
+        print("Original workspace details:")
+        pprint.pprint(workspace.model_dump())
+
+        update_data = WorkspaceUpdate(name="updated workspace", planId=8)
+
+        print(f"\n--- Updating workspace name to '{update_data.name}' ---")
+
+        await workspace.update(data=update_data)
+
+        print("\n--- Workspace successfully updated. New details: ---")
+        pprint.pprint(workspace.model_dump())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "codesphere"
 
-version = "0.1.1"
+version = "0.2.0"
 description = "Use Codesphere within python scripts."
 readme = "README.md"
 license = { file="LICENSE" }

src/codesphere/__init__.py

@@ -1,24 +1,43 @@
 # z.B. in src/codesphere/__init__.py oder einer eigenen client.py
 from .client import APIHttpClient
-from .resources.team.resource import TeamsResource
+from .resources.team.resources import TeamsResource
+from .resources.workspace.resources import WorkspacesResource
+from .resources.workspace.models import (
+    Workspace,
+    WorkspaceCreate,
+    WorkspaceUpdate,
+    WorkspaceStatus,
+)
 
 
 class CodesphereSDK:
     def __init__(self, token: str = None):
         self._http_client = APIHttpClient()
-        # Die Ressourcen werden erst im __aenter__ initialisiert
         self.teams: TeamsResource | None = None
+        self.workspaces: WorkspacesResource | None = None
 
     async def __aenter__(self):
         """Wird beim Eintritt in den 'async with'-Block aufgerufen."""
-        # Startet den internen HTTP-Client
         await self._http_client.__aenter__()
 
-        # Initialisiert die Ressourcen-Handler mit dem aktiven Client
         self.teams = TeamsResource(self._http_client)
+        self.workspaces = WorkspacesResource(self._http_client)
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         """Wird beim Verlassen des 'async with'-Blocks aufgerufen."""
-        # Schließt den internen HTTP-Client sicher
         await self._http_client.__aexit__(exc_type, exc_val, exc_tb)
+
+
+__all__ = [
+    "CodesphereSDK",
+    "CodesphereError",
+    "AuthenticationError",
+    "Team",
+    "TeamCreate",
+    "TeamInList",
+    "Workspace",
+    "WorkspaceCreate",
+    "WorkspaceUpdate",
+    "WorkspaceStatus",
+]

src/codesphere/client.py

@@ -2,6 +2,7 @@
 import httpx
 from pydantic import BaseModel
 from typing import Optional, Any
+from functools import partial
 
 from .resources.exceptions.exceptions import AuthenticationError
 
@@ -15,7 +16,11 @@ def __init__(self, base_url: str = "https://codesphere.com/api"):
 
         self._token = auth_token
         self._base_url = base_url
-        self.client: Optional[httpx.Client] = None
+        self.client: Optional[httpx.AsyncClient] = None
+
+        # Dynamically create get, post, put, patch, delete methods
+        for method in ["get", "post", "put", "patch", "delete"]:
+            setattr(self, method, partial(self.request, method.upper()))
 
     async def __aenter__(self):
         self.client = httpx.AsyncClient(
@@ -31,28 +36,16 @@ async def request(
         self, method: str, endpoint: str, **kwargs: Any
     ) -> httpx.Response:
         if not self.client:
-            raise RuntimeError("APIHttpClient must be used within a 'with' statement.")
+            raise RuntimeError(
+                "APIHttpClient must be used within an 'async with' statement."
+            )
+
+        # If a 'json' payload is a Pydantic model, automatically convert it.
+        if "json" in kwargs and isinstance(kwargs["json"], BaseModel):
+            kwargs["json"] = kwargs["json"].model_dump(exclude_none=True)
+
         print(f"{method} {endpoint} {kwargs}")
 
         response = await self.client.request(method, endpoint, **kwargs)
         response.raise_for_status()
         return response
-
-    async def get(self, endpoint: str, json: Optional[dict] = None) -> httpx.Response:
-        json_data = json.model_dump() if json else None
-        return await self.request("GET", endpoint, json=json_data)
-
-    async def post(
-        self, endpoint: str, json: Optional[BaseModel] = None
-    ) -> httpx.Response:
-        json_data = json.model_dump() if json else None
-        return await self.request("POST", endpoint, json=json_data)
-
-    async def put(
-        self, endpoint: str, json: Optional[BaseModel] = None
-    ) -> httpx.Response:
-        json_data = json.model_dump() if json else None
-        return await self.request("PUT", endpoint, json=json_data)
-
-    async def delete(self, endpoint: str) -> httpx.Response:
-        return await self.request("DELETE", endpoint)

src/codesphere/resources/base.py

@@ -39,11 +39,14 @@ async def _execute_operation(self, operation: APIOperation, **kwargs: Any) -> An
         endpoint = operation.endpoint_template.format(**format_args)
 
         params = kwargs.get("params")
+        json_data_obj = kwargs.get("data")
         payload = None
 
-        if operation.input_model:
+        if json_data_obj and isinstance(json_data_obj, BaseModel):
+            payload = json_data_obj.model_dump(exclude_none=True)
+        elif operation.input_model:
             input_data = operation.input_model(**kwargs)
-            payload = input_data.model_dump()
+            payload = input_data.model_dump(exclude_none=True)
 
         response = await self._http_client.request(
             method=operation.method, endpoint=endpoint, json=payload, params=params
@@ -54,6 +57,10 @@ async def _execute_operation(self, operation: APIOperation, **kwargs: Any) -> An
 
         json_response = response.json()
 
+        # print("--- RAW API RESPONSE ---")
+        # pprint.pprint(json_response)
+        # print("------------------------")
+
         origin = get_origin(operation.response_model)
         if origin is list or origin is List:
             item_model = get_args(operation.response_model)[0]

src/codesphere/resources/domain/models.py

@@ -1,37 +0,0 @@
-from __future__ import annotations
-from pydantic import BaseModel, PrivateAttr
-from typing import Optional, TYPE_CHECKING
-from datetime import datetime
-
-if TYPE_CHECKING:
-    from ...client import APIHttpClient
-
-
-class Domain(BaseModel):
-    _http_client: Optional[APIHttpClient] = PrivateAttr(default=None)
-
-    id: str
-    name: str
-    verified_at: Optional[datetime] = None
-
-    async def delete(self) -> None:
-        """Deletes this specific domain instance via the API."""
-        if not self._http_client:
-            raise RuntimeError("Cannot make API calls on a detached model.")
-        if not self.id:
-            raise ValueError("Cannot delete a domain without an ID.")
-
-        await self._http_client.delete(f"/domains/{self.id}")
-        print(f"Domain '{self.name}' has been deleted.")
-
-    async def save(self) -> None:
-        """Updates this domain with its current data."""
-        if not self._http_client:
-            raise RuntimeError("Cannot make API calls on a detached model.")
-        if not self.id:
-            raise ValueError("Cannot update a domain without an ID.")
-
-        update_data = self.model_dump(exclude={"id"})
-
-        await self._http_client.put(f"/domains/{self.id}", json=update_data)
-        print(f"Domain '{self.name}' has been updated.")

src/codesphere/resources/team/models.py

@@ -1,9 +1,7 @@
 from __future__ import annotations
-from pydantic import BaseModel, PrivateAttr, model_validator
+from pydantic import BaseModel, PrivateAttr
 from typing import Optional, TYPE_CHECKING
 
-from ..domain.resource import DomainsResource
-
 if TYPE_CHECKING:
     from ...client import APIHttpClient
 
@@ -41,16 +39,6 @@ class Team(TeamBase):
     """
 
     _http_client: Optional[APIHttpClient] = PrivateAttr(default=None)
-    _domains: Optional[DomainsResource] = PrivateAttr(default=None)
-
-    @model_validator(mode="after")
-    def setup_sub_resources(self) -> "Team":
-        """Creates the sub-resources after initialization."""
-        if self._http_client:
-            self._domains = DomainsResource(
-                http_client=self._http_client, team_id=str(self.id)
-            )
-        return self
 
     async def delete(self) -> None:
         """Deletes this team via the API."""