feat(workspace): impleöment base operations for workspaces

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__":

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.")
 
 

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__":

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__":

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__":

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())

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())

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__":

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())

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__":

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",

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}