Feature/envvars (#5)

* feat(src/codesphere/resources/workspace/env-vars): support env-vars endpoints of public api

* bump: version 0.3.0 → 0.4.0

---------

Co-authored-by: JD <>

Author: Datata1

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v0.4.0 (2025-07-22)
+
+### Feat
+
+- **src/codesphere/resources/workspace/env-vars**: support env-vars endpoints of public api
+
 ## v0.3.0 (2025-07-22)
 
 ### Feat

examples/workspaces/env-vars/delete_envvars.py

@@ -0,0 +1,25 @@
+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)
+
+        workspace = workspaces[0]
+
+        envs = await workspace.get_env_vars()
+        print("Current Environment Variables:")
+        pprint.pprint(envs[0].name)
+
+        await workspace.delete_env_vars([envs[0].name])  # you can pass a list of strings to delete multiple env vars
+
+        print("Environment Variables after deletion:")
+        updated_envs = await workspace.get_env_vars()
+        pprint.pprint(updated_envs)
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/workspaces/env-vars/list_envvars.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)
+
+        workspace = workspaces[0]
+
+        envs = await workspace.get_env_vars()
+        print("Current Environment Variables:")
+        pprint.pprint(envs)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/workspaces/env-vars/set_envvars.py

@@ -0,0 +1,27 @@
+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)
+
+        workspace = workspaces[0]
+
+        envs = await workspace.get_env_vars()
+        print("Current Environment Variables:")
+        pprint.pprint(envs)
+
+        envs[0].value = "new_value"  # Modify an environment variable
+        await workspace.set_env_vars(envs)  # Update the environment variables
+
+        print("Updated Environment Variables:")
+        updated_envs = await workspace.get_env_vars()
+        pprint.pprint(updated_envs)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

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

src/codesphere/client.py

@@ -23,8 +23,11 @@ def __init__(self, base_url: str = "https://codesphere.com/api"):
             setattr(self, method, partial(self.request, method.upper()))
 
     async def __aenter__(self):
+        timeout_config = httpx.Timeout(10.0, read=30.0)
         self.client = httpx.AsyncClient(
-            base_url=self._base_url, headers={"Authorization": f"Bearer {self._token}"}
+            base_url=self._base_url,
+            headers={"Authorization": f"Bearer {self._token}"},
+            timeout=timeout_config,
         )
         return self
 

src/codesphere/resources/workspace/env-vars/model.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+from pydantic import BaseModel, PrivateAttr
+from typing import Optional, List, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ...client import APIHttpClient
+
+
+class EnvVarPair(BaseModel):
+    name: str
+    value: str
+
+
+class WorkspaceCreate(BaseModel):
+    teamId: int
+    name: str
+    planId: int
+    baseImage: Optional[str] = None
+    isPrivateRepo: bool = True
+    replicas: int = 1
+    gitUrl: Optional[str] = None
+    initialBranch: Optional[str] = None
+    cloneDepth: Optional[int] = None
+    sourceWorkspaceId: Optional[int] = None
+    welcomeMessage: Optional[str] = None
+    vpnConfig: Optional[str] = None
+    restricted: Optional[bool] = None
+    env: Optional[List[EnvVarPair]] = None
+
+
+# Defines the request body for PATCH /workspaces/{workspaceId}
+class WorkspaceUpdate(BaseModel):
+    planId: Optional[int] = None
+    baseImage: Optional[str] = None
+    name: Optional[str] = None
+    replicas: Optional[int] = None
+    vpnConfig: Optional[str] = None
+    restricted: Optional[bool] = None
+
+
+# Defines the response from GET /workspaces/{workspaceId}/status
+class WorkspaceStatus(BaseModel):
+    isRunning: bool
+
+
+# This is the main model for a workspace, returned by GET, POST, and LIST
+class Workspace(BaseModel):
+    _http_client: Optional[APIHttpClient] = PrivateAttr(default=None)
+
+    id: int
+    teamId: int
+    name: str
+    planId: int
+    isPrivateRepo: bool
+    replicas: int
+    baseImage: Optional[str] = None
+    dataCenterId: int
+    userId: int
+    gitUrl: Optional[str] = None
+    initialBranch: Optional[str] = None
+    sourceWorkspaceId: Optional[int] = None
+    welcomeMessage: Optional[str] = None
+    vpnConfig: Optional[str] = None
+    restricted: bool
+
+    async def update(self, data: WorkspaceUpdate) -> None:
+        """Updates this workspace with new data."""
+        if not self._http_client:
+            raise RuntimeError("Cannot make API calls on a detached model.")
+
+        await self._http_client.patch(
+            f"/workspaces/{self.id}", json=data.model_dump(exclude_unset=True)
+        )
+        # Optionally, update the local object's state
+        for key, value in data.model_dump(exclude_unset=True).items():
+            setattr(self, key, value)
+
+    async def delete(self) -> None:
+        """Deletes this workspace."""
+        if not self._http_client:
+            raise RuntimeError("Cannot make API calls on a detached model.")
+        await self._http_client.delete(f"/workspaces/{self.id}")
+
+    async def get_status(self) -> WorkspaceStatus:
+        """Gets the running status of this workspace."""
+        if not self._http_client:
+            raise RuntimeError("Cannot make API calls on a detached model.")
+
+        response = await self._http_client.get(f"/workspaces/{self.id}/status")
+        return WorkspaceStatus.model_validate(response.json())

src/codesphere/resources/workspace/env-vars/modely.py

@@ -0,0 +1,39 @@
+from typing import List
+from ..base import ResourceBase, APIOperation
+from .models import Workspace, WorkspaceCreate, WorkspaceUpdate
+
+
+class WorkspacesResource(ResourceBase):
+    """Manages all API operations for the Workspace resource."""
+
+    list_by_team = APIOperation(
+        method="GET",
+        endpoint_template="/workspaces/team/{team_id}",
+        response_model=List[Workspace],
+    )
+
+    get = APIOperation(
+        method="GET",
+        endpoint_template="/workspaces/{workspace_id}",
+        response_model=Workspace,
+    )
+
+    create = APIOperation(
+        method="POST",
+        endpoint_template="/workspaces",
+        input_model=WorkspaceCreate,
+        response_model=Workspace,
+    )
+
+    update = APIOperation(
+        method="PATCH",
+        endpoint_template="/workspaces/{workspace_id}",
+        input_model=WorkspaceUpdate,
+        response_model=None,
+    )
+
+    delete = APIOperation(
+        method="DELETE",
+        endpoint_template="/workspaces/{workspace_id}",
+        response_model=None,
+    )

src/codesphere/resources/workspace/env-vars/resources.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
-from pydantic import BaseModel, PrivateAttr
-from typing import Optional, List, TYPE_CHECKING
+from pydantic import BaseModel, PrivateAttr, parse_obj_as
+from typing import Optional, List, TYPE_CHECKING, Union, Dict
 
 if TYPE_CHECKING:
     from ...client import APIHttpClient
@@ -88,3 +88,47 @@ async def get_status(self) -> WorkspaceStatus:
 
         response = await self._http_client.get(f"/workspaces/{self.id}/status")
         return WorkspaceStatus.model_validate(response.json())
+
+    async def get_env_vars(self) -> list[EnvVarPair]:
+        """Fetches all environment variables for this workspace."""
+        if not self._http_client:
+            raise RuntimeError("Cannot make API calls on a detached model.")
+
+        response = await self._http_client.get(f"/workspaces/{self.id}/env-vars")
+        return parse_obj_as(list[EnvVarPair], response.json())
+
+    async def set_env_vars(
+        self, env_vars: Union[List[EnvVarPair], List[Dict[str, str]]]
+    ) -> None:
+        """
+        Sets or updates environment variables for this workspace.
+        This operation replaces all existing variables with the provided list.
+        Accepts either a list of EnvVarPair models or a list of dictionaries.
+        """
+        if not self._http_client:
+            raise RuntimeError("Cannot make API calls on a detached model.")
+
+        json_payload = []
+        if env_vars and isinstance(env_vars[0], EnvVarPair):
+            json_payload = [var.model_dump() for var in env_vars]
+        else:
+            json_payload = env_vars
+
+        await self._http_client.put(
+            f"/workspaces/{self.id}/env-vars", json=json_payload
+        )
+
+    async def delete_env_vars(
+        self, var_names: Union[List[str], List[EnvVarPair]]
+    ) -> None:
+        """Deletes specific environment variables from this workspace."""
+        if not self._http_client:
+            raise RuntimeError("Cannot make API calls on a detached model.")
+
+        payload = []
+        if var_names and isinstance(var_names[0], EnvVarPair):
+            payload = [var.name for var in var_names]
+        else:
+            payload = var_names
+
+        await self._http_client.delete(f"/workspaces/{self.id}/env-vars", json=payload)