From 0f8d7dff0fae7a7f51d4e370c93b79357a84e585 Mon Sep 17 00:00:00 2001 From: Datata1 <> Date: Sat, 8 Nov 2025 23:50:21 +0100 Subject: [PATCH] =?UTF-8?q?feat(workspace):=20imple=C3=B6ment=20base=20ope?= =?UTF-8?q?rations=20for=20workspaces?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- examples/workspaces/create_workspace.py | 14 +- examples/workspaces/delete_workspace.py | 9 +- .../workspaces/env-vars/delete_envvars.py | 23 +- examples/workspaces/env-vars/list_envvars.py | 10 +- examples/workspaces/env-vars/set_envvars.py | 25 +-- examples/workspaces/execute_command.py | 32 +++ examples/workspaces/get_status.py | 19 ++ examples/workspaces/get_workspace.py | 21 +- examples/workspaces/list_workspace.py | 22 ++ examples/workspaces/list_workspaces.py | 20 -- examples/workspaces/update_workspace.py | 20 +- src/codesphere/__init__.py | 2 + src/codesphere/core/handler.py | 18 +- .../resources/workspace/env-vars/models.py | 90 -------- .../resources/workspace/env-vars/resources.py | 39 ---- .../resources/workspace/envVars/__init__.py | 3 + .../resources/workspace/envVars/models.py | 81 +++++++ src/codesphere/resources/workspace/models.py | 198 ++++++++++-------- 18 files changed, 350 insertions(+), 296 deletions(-) create mode 100644 examples/workspaces/get_status.py create mode 100644 examples/workspaces/list_workspace.py delete mode 100644 examples/workspaces/list_workspaces.py delete mode 100644 src/codesphere/resources/workspace/env-vars/models.py delete mode 100644 src/codesphere/resources/workspace/env-vars/resources.py create mode 100644 src/codesphere/resources/workspace/envVars/__init__.py create mode 100644 src/codesphere/resources/workspace/envVars/models.py diff --git a/examples/workspaces/create_workspace.py b/examples/workspaces/create_workspace.py index 53c9ce8..1d17d1d 100644 --- a/examples/workspaces/create_workspace.py +++ b/examples/workspaces/create_workspace.py @@ -1,27 +1,25 @@ import asyncio -import pprint +import logging from codesphere import CodesphereSDK, WorkspaceCreate +logging.basicConfig(level=logging.INFO) + async def main(): - """Creates a new workspace in a specific team.""" - team_id = 12345 + team_id = int(999999) 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), + teamId=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_json()) + print(created_workspace.model_dump_json()) if __name__ == "__main__": diff --git a/examples/workspaces/delete_workspace.py b/examples/workspaces/delete_workspace.py index 6a4b6c7..aeba2ce 100644 --- a/examples/workspaces/delete_workspace.py +++ b/examples/workspaces/delete_workspace.py @@ -1,23 +1,22 @@ import asyncio +import logging from codesphere import CodesphereSDK +logging.basicConfig(level=logging.INFO) + async def main(): """Deletes a specific workspace.""" - workspace_id_to_delete = 12345 + workspace_id_to_delete = int(9999999) 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.") diff --git a/examples/workspaces/env-vars/delete_envvars.py b/examples/workspaces/env-vars/delete_envvars.py index fc3273f..260e6ee 100644 --- a/examples/workspaces/env-vars/delete_envvars.py +++ b/examples/workspaces/env-vars/delete_envvars.py @@ -1,27 +1,26 @@ import asyncio -import pprint +import logging from codesphere import CodesphereSDK +logging.basicConfig(level=logging.INFO) + 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] + vars_to_delete = await workspace.env_vars.get() + for env in vars_to_delete: + print(env.model_dump_json(indent=2)) - 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 + await workspace.env_vars.delete(vars_to_delete) - print("Environment Variables after deletion:") - updated_envs = await workspace.get_env_vars() - pprint.pprint(updated_envs) + print("\n--- Verifying deletion ---") + current_vars = await workspace.env_vars.get() + for env in current_vars: + print(env.model_dump_json(indent=2)) if __name__ == "__main__": diff --git a/examples/workspaces/env-vars/list_envvars.py b/examples/workspaces/env-vars/list_envvars.py index e0719f3..3428371 100644 --- a/examples/workspaces/env-vars/list_envvars.py +++ b/examples/workspaces/env-vars/list_envvars.py @@ -1,19 +1,21 @@ import asyncio -import pprint +import logging from codesphere import CodesphereSDK +logging.basicConfig(level=logging.INFO) + 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() + envs = await workspace.env_vars.get() print("Current Environment Variables:") - pprint.pprint(envs) + for env in envs: + print(env.model_dump_json(indent=2)) if __name__ == "__main__": diff --git a/examples/workspaces/env-vars/set_envvars.py b/examples/workspaces/env-vars/set_envvars.py index 2ff8f7f..c9a6c4d 100644 --- a/examples/workspaces/env-vars/set_envvars.py +++ b/examples/workspaces/env-vars/set_envvars.py @@ -1,26 +1,27 @@ import asyncio -import pprint -from codesphere import CodesphereSDK +import logging +from codesphere import CodesphereSDK, EnvVar + +logging.basicConfig(level=logging.INFO) 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) + new_vars = [ + EnvVar(name="MY_NEW_VAR", value="hello_world"), + EnvVar(name="ANOTHER_VAR", value="123456"), + ] - envs[0].value = "new_value" # Modify an environment variable - await workspace.set_env_vars(envs) # Update the environment variables + await workspace.env_vars.set(new_vars) - print("Updated Environment Variables:") - updated_envs = await workspace.get_env_vars() - pprint.pprint(updated_envs) + print("\n--- Verifying new list ---") + current_vars = await workspace.env_vars.get() + for env in current_vars: + print(env.model_dump_json(indent=2)) if __name__ == "__main__": diff --git a/examples/workspaces/execute_command.py b/examples/workspaces/execute_command.py index e69de29..fdae358 100644 --- a/examples/workspaces/execute_command.py +++ b/examples/workspaces/execute_command.py @@ -0,0 +1,32 @@ +import asyncio +import logging +from codesphere import CodesphereSDK + +# --- Logging-Konfiguration --- +logging.basicConfig(level=logging.INFO) +# (Optionale Logger stummschalten) +logging.getLogger("codesphere.http_client").setLevel(logging.WARNING) +logging.getLogger("httpx").setLevel(logging.WARNING) + +log = logging.getLogger(__name__) + + +async def main(): + 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] + state = await workspace.get_status() + print(state.model_dump_json(indent=2)) + + command_str = "echo Hello from $USER_NAME!" + command_env = {"USER_NAME": "SDK-User"} + + command_output = await workspace.execute_command( + command=command_str, env=command_env + ) + print(command_output.model_dump_json(indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/workspaces/get_status.py b/examples/workspaces/get_status.py new file mode 100644 index 0000000..a9c631c --- /dev/null +++ b/examples/workspaces/get_status.py @@ -0,0 +1,19 @@ +import asyncio +import logging +from codesphere import CodesphereSDK + +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + all_teams = await sdk.teams.list() + first_team = all_teams[0] + workspaces = await sdk.workspaces.list_by_team(team_id=first_team.id) + first_workspace = workspaces[0] + state = await first_workspace.get_status() + print(state.model_dump_json(indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/workspaces/get_workspace.py b/examples/workspaces/get_workspace.py index e950243..8ddc29c 100644 --- a/examples/workspaces/get_workspace.py +++ b/examples/workspaces/get_workspace.py @@ -1,11 +1,28 @@ import asyncio +import logging from codesphere import CodesphereSDK +logging.basicConfig(level=logging.INFO) + async def main(): - """Fetches a workspace within a Team.""" async with CodesphereSDK() as sdk: - pass + all_teams = await sdk.teams.list() + if not all_teams: + print("No teams found. Cannot get a workspace.") + return + + first_team = all_teams[0] + + workspaces = await sdk.workspaces.list_by_team(team_id=first_team.id) + if not workspaces: + print(f"No workspaces found in team '{first_team.name}'.") + return + + first_workspace = workspaces[0] + workspace_id_to_fetch = first_workspace.id + workspace = await sdk.workspaces.get(workspace_id=workspace_id_to_fetch) + print(workspace.model_dump_json(indent=2)) if __name__ == "__main__": diff --git a/examples/workspaces/list_workspace.py b/examples/workspaces/list_workspace.py new file mode 100644 index 0000000..387a4bd --- /dev/null +++ b/examples/workspaces/list_workspace.py @@ -0,0 +1,22 @@ +import asyncio +import logging +from codesphere import CodesphereSDK + +# --- Logging-Konfiguration --- +logging.basicConfig(level=logging.INFO) + + +async def main(): + async with CodesphereSDK() as sdk: + all_teams = await sdk.teams.list() + first_team = all_teams[0] + team_id_to_fetch = first_team.id + workspaces = await sdk.workspaces.list_by_team(team_id=team_id_to_fetch) + print(f"\n--- Workspaces in Team: {first_team.name} ---") + for ws in workspaces: + print(ws.model_dump_json(indent=2)) + print("-" * 20) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/workspaces/list_workspaces.py b/examples/workspaces/list_workspaces.py deleted file mode 100644 index 8e1e6cb..0000000 --- a/examples/workspaces/list_workspaces.py +++ /dev/null @@ -1,20 +0,0 @@ -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()) diff --git a/examples/workspaces/update_workspace.py b/examples/workspaces/update_workspace.py index 19dd56f..ed951d2 100644 --- a/examples/workspaces/update_workspace.py +++ b/examples/workspaces/update_workspace.py @@ -1,27 +1,21 @@ import asyncio -import pprint +import logging from codesphere import CodesphereSDK, WorkspaceUpdate +logging.basicConfig(level=logging.INFO) + async def main(): """Fetches a workspace and updates its name.""" - workspace_id_to_update = 12245 + workspace_id_to_update = 72678 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(workspace.model_dump_json(indent=2)) - 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}' ---") - + update_data = WorkspaceUpdate(name="updated workspace2", planId=8) await workspace.update(data=update_data) - - print("\n--- Workspace successfully updated. New details: ---") - pprint.pprint(workspace.model_dump()) + print(workspace.model_dump_json(indent=2)) if __name__ == "__main__": diff --git a/src/codesphere/__init__.py b/src/codesphere/__init__.py index c4f9eca..a09318a 100644 --- a/src/codesphere/__init__.py +++ b/src/codesphere/__init__.py @@ -30,6 +30,7 @@ WorkspaceUpdate, WorkspaceStatus, ) +from .resources.workspace.envVars import EnvVar from .resources.metadata import Datacenter, Characteristic, WsPlan, Image logging.getLogger("codesphere").addHandler(logging.NullHandler()) @@ -45,6 +46,7 @@ "WorkspaceCreate", "WorkspaceUpdate", "WorkspaceStatus", + "EnvVar", "Datacenter", "Characteristic", "WsPlan", diff --git a/src/codesphere/core/handler.py b/src/codesphere/core/handler.py index 397149c..5d7f31e 100644 --- a/src/codesphere/core/handler.py +++ b/src/codesphere/core/handler.py @@ -4,6 +4,7 @@ import httpx from pydantic import BaseModel, PrivateAttr, ValidationError +from pydantic.fields import FieldInfo from ..http_client import APIHttpClient from .operations import APIOperation @@ -16,9 +17,17 @@ class _APIOperationExecutor: def __getattribute__(self, name: str) -> Any: attr = super().__getattribute__(name) + operation = None + + if isinstance(attr, FieldInfo): + if isinstance(attr.default, APIOperation): + operation = attr.default + elif isinstance(attr, APIOperation): + operation = attr + + if operation: + return partial(self._execute_operation, operation=operation) - if isinstance(attr, APIOperation): - return partial(self._execute_operation, operation=attr) return attr async def _execute_operation(self, operation: APIOperation, **kwargs: Any) -> Any: @@ -48,15 +57,20 @@ async def execute(self) -> Any: def _prepare_request_args(self) -> tuple[str, dict]: format_args = {} + format_args.update(self.kwargs) + format_args.update(self.executor.__dict__) if isinstance(self.executor, BaseModel): format_args.update(self.executor.model_dump()) format_args.update(self.kwargs) + endpoint = self.operation.endpoint_template.format(**format_args) payload = None if json_data_obj := self.kwargs.get("data"): if isinstance(json_data_obj, BaseModel): payload = json_data_obj.model_dump(exclude_none=True) + else: + payload = json_data_obj request_kwargs = {"params": self.kwargs.get("params"), "json": payload} return endpoint, {k: v for k, v in request_kwargs.items() if v is not None} diff --git a/src/codesphere/resources/workspace/env-vars/models.py b/src/codesphere/resources/workspace/env-vars/models.py deleted file mode 100644 index 4a7d1fb..0000000 --- a/src/codesphere/resources/workspace/env-vars/models.py +++ /dev/null @@ -1,90 +0,0 @@ -from __future__ import annotations -from pydantic import BaseModel, PrivateAttr -from typing import Optional, List, TYPE_CHECKING - -if TYPE_CHECKING: - from ....http_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()) diff --git a/src/codesphere/resources/workspace/env-vars/resources.py b/src/codesphere/resources/workspace/env-vars/resources.py deleted file mode 100644 index 6a44930..0000000 --- a/src/codesphere/resources/workspace/env-vars/resources.py +++ /dev/null @@ -1,39 +0,0 @@ -from typing import List -from ....core 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, - ) diff --git a/src/codesphere/resources/workspace/envVars/__init__.py b/src/codesphere/resources/workspace/envVars/__init__.py new file mode 100644 index 0000000..3a9791e --- /dev/null +++ b/src/codesphere/resources/workspace/envVars/__init__.py @@ -0,0 +1,3 @@ +from .models import EnvVar, WorkspaceEnvVarManager + +__all__ = ["EnvVar", "WorkspaceEnvVarManager"] diff --git a/src/codesphere/resources/workspace/envVars/models.py b/src/codesphere/resources/workspace/envVars/models.py new file mode 100644 index 0000000..9f6b408 --- /dev/null +++ b/src/codesphere/resources/workspace/envVars/models.py @@ -0,0 +1,81 @@ +from __future__ import annotations +import logging +from typing import Dict, List, Union +from pydantic import BaseModel, Field + +from codesphere.core.handler import _APIOperationExecutor +from codesphere.core.operations import APIOperation, AsyncCallable +from codesphere.http_client import APIHttpClient + +log = logging.getLogger(__name__) + + +class EnvVar(BaseModel): + name: str + value: str + + +class WorkspaceEnvVarManager(_APIOperationExecutor): + """ + Verwaltet die Env Vars für einen *bestimmten* Workspace. + Wird typischerweise über 'workspace.env_vars' aufgerufen. + """ + + def __init__(self, http_client: APIHttpClient, workspace_id: int): + self._http_client = http_client + self._workspace_id = workspace_id + self.id = workspace_id + + get_all_op: AsyncCallable[List[EnvVar]] = Field( + default=APIOperation( + method="GET", + endpoint_template="/workspaces/{id}/env-vars", + response_model=List[EnvVar], + ), + exclude=True, + ) + + bulk_set_op: AsyncCallable[None] = Field( + default=APIOperation( + method="PUT", + endpoint_template="/workspaces/{id}/env-vars", + response_model=None, + ), + exclude=True, + ) + + bulk_delete_op: AsyncCallable[None] = Field( + default=APIOperation( + method="DELETE", + endpoint_template="/workspaces/{id}/env-vars", + response_model=None, + ), + exclude=True, + ) + + async def get(self) -> List[EnvVar]: + """Fetches all environment variables for this workspace.""" + env_vars = await self.get_all_op() + return env_vars + + async def set(self, env_vars: Union[List[EnvVar], List[Dict[str, str]]]) -> None: + """Sets or updates environment variables for this workspace.""" + json_payload = [] + if env_vars: + if isinstance(env_vars[0], EnvVar): + json_payload = [var.model_dump() for var in env_vars] + else: + json_payload = env_vars + + await self.bulk_set_op(data=json_payload) + + async def delete(self, var_names: Union[List[str], List[EnvVar]]) -> None: + """Deletes specific environment variables from this workspace.""" + payload = [] + if var_names: + if isinstance(var_names[0], EnvVar): + payload = [var.name for var in var_names] + else: + payload = var_names + + await self.bulk_delete_op(data=payload) diff --git a/src/codesphere/resources/workspace/models.py b/src/codesphere/resources/workspace/models.py index a2e6362..deb8534 100644 --- a/src/codesphere/resources/workspace/models.py +++ b/src/codesphere/resources/workspace/models.py @@ -1,14 +1,13 @@ from __future__ import annotations -from pydantic import BaseModel, PrivateAttr, parse_obj_as -from typing import Optional, List, TYPE_CHECKING, Union, Dict +from functools import cached_property +import logging +from pydantic import BaseModel, Field +from typing import Dict, Optional, List -if TYPE_CHECKING: - from ...http_client import APIHttpClient +from ...core import _APIOperationExecutor, APIOperation, AsyncCallable +from .envVars import EnvVar, WorkspaceEnvVarManager - -class EnvVarPair(BaseModel): - name: str - value: str +log = logging.getLogger(__name__) class WorkspaceCreate(BaseModel): @@ -25,28 +24,10 @@ class WorkspaceCreate(BaseModel): welcomeMessage: Optional[str] = None vpnConfig: Optional[str] = None restricted: Optional[bool] = None - env: Optional[List[EnvVarPair]] = None + env: Optional[List[EnvVar]] = 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) - +class WorkspaceBase(BaseModel): id: int teamId: int name: str @@ -63,81 +44,120 @@ class Workspace(BaseModel): vpnConfig: Optional[str] = None restricted: bool + +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 + + +class WorkspaceStatus(BaseModel): + isRunning: bool + + +class CommandInput(BaseModel): + command: str + env: Optional[Dict[str, str]] = None + + +class CommandOutput(BaseModel): + command: str + workingDir: str + output: str + error: str + + +class Workspace(WorkspaceBase, _APIOperationExecutor): + update_op: AsyncCallable[None] = Field( + default=APIOperation( + method="PATCH", + endpoint_template="/workspaces/{id}", + response_model=None, + ), + exclude=True, + ) + + delete_op: AsyncCallable[None] = Field( + default=APIOperation( + method="DELETE", + endpoint_template="/workspaces/{id}", + response_model=None, + ), + exclude=True, + ) + + get_status_op: AsyncCallable[WorkspaceStatus] = Field( + default=APIOperation( + method="GET", + endpoint_template="/workspaces/{id}/status", + response_model=WorkspaceStatus, + ), + exclude=True, + ) + + execute_command_op: AsyncCallable[CommandOutput] = Field( + default=APIOperation( + method="POST", + endpoint_template="/workspaces/{id}/execute", + input_model=CommandInput, + response_model=CommandOutput, + ), + exclude=True, + ) + 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.") + """ + Updates this workspace with new data and refreshes the + local object state. - 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(): + Args: + data (WorkspaceUpdate): The payload with fields to update. + """ + await self.update_op(data=data) + update_data = data.model_dump(exclude_unset=True) + log.debug(f"Updating local workspace state (id={self.id}) with: {update_data}") + for key, value in update_data.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}") + await self.delete_op() 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.") + return await self.get_status_op() - response = await self._http_client.get(f"/workspaces/{self.id}/status") - return WorkspaceStatus.model_validate(response.json()) + async def execute_command( + self, command: str, env: Optional[Dict[str, str]] = None + ) -> CommandOutput: + """ + Führt einen Befehl in diesem Workspace aus. - 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.") + Args: + command (str): Der Bash-Befehl (z.B. "ls -la"). + env (Dict[str, str], optional): Env Vars, die nur + für diesen Befehl gesetzt werden. - response = await self._http_client.get(f"/workspaces/{self.id}/env-vars") - return parse_obj_as(list[EnvVarPair], response.json()) + Returns: + CommandOutput: Ein Objekt mit stdout und stderr. + """ + command_data = CommandInput(command=command, env=env) + return await self.execute_command_op(data=command_data) - async def set_env_vars( - self, env_vars: Union[List[EnvVarPair], List[Dict[str, str]]] - ) -> None: + @cached_property + def env_vars(self) -> WorkspaceEnvVarManager: """ - 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. + Provides access to the Environment Variable manager for this workspace. + + Usage: + >>> await workspace.env_vars.get() + >>> await workspace.env_vars.set([{"name": "KEY", "value": "VALUE"}]) """ 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 + raise RuntimeError("Cannot access 'env_vars' on a detached model.") + return WorkspaceEnvVarManager( + http_client=self._http_client, workspace_id=self.id ) - - 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) - - async def execute_command(): - pass - - async def git_pull(): - pass - - async def git_head(): - pass