context manager for log streams

Author: Datata1

examples/create_workspace_with_landscape.py

@@ -1,7 +1,3 @@
-"""
-Demo: Create a workspace, deploy a landscape profile, and stream logs.
-"""
-
 import asyncio
 import time
 
@@ -14,7 +10,7 @@
 )
 from codesphere.resources.workspace.logs import LogStage
 
-TEAM_ID = 35698
+TEAM_ID = 123
 
 
 async def main():
@@ -25,8 +21,8 @@ async def main():
             raise ValueError("Micro plan not found")
 
         workspace_name = f"pipeline-demo-{int(time.time())}"
-        print(f"Creating workspace '{workspace_name}'...")
 
+        print(f"Creating workspace '{workspace_name}'...")
         workspace = await sdk.workspaces.create(
             WorkspaceCreate(plan_id=plan.id, team_id=TEAM_ID, name=workspace_name)
         )
@@ -45,7 +41,7 @@ async def main():
             .add_reactive_service("web")
             .plan(plan.id)
             .add_step(
-                'for i in $(seq 1 50); do echo "[$i] Processing request..."; sleep 1; done'
+                'for i in $(seq 1 20); do echo "[$i] Processing request..."; sleep 1; done'
             )
             .add_port(3000, public=True)
             .add_path("/", port=3000)
@@ -82,17 +78,17 @@ async def main():
 
         print("\n--- Run Stage ---")
         await workspace.landscape.start_stage(PipelineStage.RUN, profile="production")
-        print("Started run stage, waiting for logs...\n")
+        print("Started run stage\n")
 
-        print("Streaming logs from 'web' service:")
+        print("Streaming logs from 'web' service (using context manager):")
         count = 0
-        async for entry in workspace.logs.stream_server(step=0, server="web"):
-            if entry.get_text():
-                print(f"  {entry.get_text().strip()}")
-                count += 1
+        async with workspace.logs.open_server_stream(step=0, server="web") as stream:
+            async for entry in stream:
+                if entry.get_text():
+                    print(f"  {entry.get_text().strip()}")
+                    count += 1
 
         print(f"\n✓ Stream ended ({count} log entries)")
-        print(f"✓ Workspace {workspace.id} is still running.")
 
 
 if __name__ == "__main__":

src/codesphere/core/__init__.py

@@ -1,11 +1,13 @@
-from .base import ResourceBase
-from .operations import APIOperation, AsyncCallable
-from .handler import _APIOperationExecutor, APIRequestHandler
+from .base import CamelModel, ResourceBase
+from .handler import APIRequestHandler, _APIOperationExecutor
+from .operations import APIOperation, AsyncCallable, StreamOperation
 
 __all__ = [
+    "CamelModel",
     "ResourceBase",
     "APIOperation",
     "_APIOperationExecutor",
     "APIRequestHandler",
     "AsyncCallable",
+    "StreamOperation",
 ]

src/codesphere/core/operations.py

@@ -1,17 +1,25 @@
-from typing import Callable, Awaitable, Generic, Optional, Type, TypeAlias, TypeVar
-
-from pydantic import BaseModel
+from typing import Awaitable, Callable, Generic, Optional, Type, TypeAlias, TypeVar
 
+from pydantic import BaseModel, ConfigDict
 
 _T = TypeVar("_T")
 ResponseT = TypeVar("ResponseT")
 InputT = TypeVar("InputT")
+EntryT = TypeVar("EntryT")
 
 AsyncCallable: TypeAlias = Callable[[], Awaitable[_T]]
 
 
 class APIOperation(BaseModel, Generic[ResponseT, InputT]):
+    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)
+
     method: str
     endpoint_template: str
     response_model: Type[ResponseT]
     input_model: Optional[Type[InputT]] = None
+
+
+class StreamOperation(BaseModel, Generic[EntryT]):
+    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)
+    endpoint_template: str
+    entry_model: Type[EntryT]

src/codesphere/resources/workspace/__init__.py

@@ -1,5 +1,5 @@
 from .git import GitHead, WorkspaceGitManager
-from .logs import LogEntry, LogProblem, LogStage, WorkspaceLogManager
+from .logs import LogEntry, LogProblem, LogStage, LogStream, WorkspaceLogManager
 from .resources import WorkspacesResource
 from .schemas import (
     CommandInput,
@@ -20,6 +20,7 @@
     "CommandOutput",
     "WorkspaceGitManager",
     "GitHead",
+    "LogStream",
     "WorkspaceLogManager",
     "LogEntry",
     "LogProblem",

src/codesphere/resources/workspace/landscape/models.py

@@ -107,24 +107,11 @@ async def teardown(self) -> None:
     async def scale(self, services: Dict[str, int]) -> None:
         await self._execute_operation(_SCALE_OP, data=services)
 
-    # Pipeline operations
-
     async def start_stage(
         self,
         stage: Union[PipelineStage, str],
         profile: Optional[str] = None,
     ) -> None:
-        """Start a pipeline stage.
-
-        Args:
-            stage: The pipeline stage to start ('prepare', 'test', or 'run').
-            profile: Optional profile name. If provided, starts the stage with
-                     that profile. Required for first run after deploy.
-
-        Raises:
-            ValidationError: If the workspace is not running or parameters are invalid.
-            NotFoundError: If the workspace is not found.
-        """
         if isinstance(stage, PipelineStage):
             stage = stage.value
 
@@ -137,15 +124,6 @@ async def start_stage(
             await self._execute_operation(_START_PIPELINE_STAGE_OP, stage=stage)
 
     async def stop_stage(self, stage: Union[PipelineStage, str]) -> None:
-        """Stop a pipeline stage.
-
-        Args:
-            stage: The pipeline stage to stop ('prepare', 'test', or 'run').
-
-        Raises:
-            ValidationError: If the workspace is not running or parameters are invalid.
-            NotFoundError: If the workspace is not found.
-        """
         if isinstance(stage, PipelineStage):
             stage = stage.value
 
@@ -154,18 +132,6 @@ async def stop_stage(self, stage: Union[PipelineStage, str]) -> None:
     async def get_stage_status(
         self, stage: Union[PipelineStage, str]
     ) -> PipelineStatusList:
-        """Get the status of a pipeline stage.
-
-        Args:
-            stage: The pipeline stage to get status for ('prepare', 'test', or 'run').
-
-        Returns:
-            List of PipelineStatus objects, one per replica/server.
-
-        Raises:
-            ValidationError: If the workspace is not running or parameters are invalid.
-            NotFoundError: If the workspace is not found.
-        """
         if isinstance(stage, PipelineStage):
             stage = stage.value
 
@@ -179,22 +145,6 @@ async def wait_for_stage(
         poll_interval: float = 5.0,
         server: Optional[str] = None,
     ) -> PipelineStatusList:
-        """Wait for a pipeline stage to complete (success or failure).
-
-        Args:
-            stage: The pipeline stage to wait for.
-            timeout: Maximum time to wait in seconds (default: 300).
-            poll_interval: Time between status checks in seconds (default: 5).
-            server: Optional server name to filter by. If None, waits for all
-                    servers that have steps defined for this stage.
-
-        Returns:
-            Final PipelineStatusList when stage completes.
-
-        Raises:
-            TimeoutError: If the stage doesn't complete within the timeout.
-            ValidationError: If the workspace is not running.
-        """
         if poll_interval <= 0:
             raise ValueError("poll_interval must be greater than 0")
 
@@ -204,26 +154,17 @@ async def wait_for_stage(
         while elapsed < timeout:
             status_list = await self.get_stage_status(stage)
 
-            # Filter to relevant servers for THIS stage
-            # A server is relevant for this stage if:
-            # - It has steps defined (meaning it participates in this stage)
-            # - OR it's not in 'waiting' state (meaning it has started)
             relevant_statuses = []
             for s in status_list:
                 if server is not None:
-                    # Filter by specific server
                     if s.server == server:
                         relevant_statuses.append(s)
                 else:
-                    # Include servers that have steps for this stage
-                    # Servers with no steps and waiting state don't participate in this stage
                     if s.steps:
                         relevant_statuses.append(s)
                     elif s.state != PipelineState.WAITING:
-                        # Started but no steps visible yet
                         relevant_statuses.append(s)
 
-            # If no relevant statuses yet, keep waiting
             if not relevant_statuses:
                 log.debug(
                     "Pipeline stage '%s': no servers with steps yet, waiting...",
@@ -233,7 +174,6 @@ async def wait_for_stage(
                 elapsed += poll_interval
                 continue
 
-            # Check if all relevant servers have completed
             all_completed = all(
                 s.state
                 in (PipelineState.SUCCESS, PipelineState.FAILURE, PipelineState.ABORTED)
@@ -244,7 +184,6 @@ async def wait_for_stage(
                 log.debug("Pipeline stage '%s' completed.", stage_name)
                 return PipelineStatusList(root=relevant_statuses)
 
-            # Log current state
             states = [f"{s.server}={s.state.value}" for s in relevant_statuses]
             log.debug(
                 "Pipeline stage '%s' status: %s (elapsed: %.1fs)",

src/codesphere/resources/workspace/landscape/operations.py

@@ -25,7 +25,6 @@
     response_model=type(None),
 )
 
-# Pipeline operations
 _START_PIPELINE_STAGE_OP = APIOperation(
     method="POST",
     endpoint_template="/workspaces/{id}/pipeline/{stage}/start",

src/codesphere/resources/workspace/landscape/schemas.py

@@ -10,16 +10,12 @@
 
 
 class PipelineStage(str, Enum):
-    """Pipeline stage for operations."""
-
     PREPARE = "prepare"
     TEST = "test"
     RUN = "run"
 
 
 class PipelineState(str, Enum):
-    """State of a pipeline stage or step."""
-
     WAITING = "waiting"
     RUNNING = "running"
     SUCCESS = "success"
@@ -28,16 +24,12 @@ class PipelineState(str, Enum):
 
 
 class StepStatus(CamelModel):
-    """Status of a single pipeline step."""
-
     state: PipelineState
     started_at: Optional[str] = None
     finished_at: Optional[str] = None
 
 
 class PipelineStatus(CamelModel):
-    """Status of a pipeline stage execution."""
-
     state: PipelineState
     started_at: Optional[str] = None
     finished_at: Optional[str] = None
@@ -47,8 +39,6 @@ class PipelineStatus(CamelModel):
 
 
 class PipelineStatusList(RootModel[List[PipelineStatus]]):
-    """List of pipeline status entries (one per replica/server)."""
-
     root: List[PipelineStatus]
 
     def __iter__(self):

src/codesphere/resources/workspace/logs/__init__.py

@@ -1,7 +1,8 @@
-from .models import WorkspaceLogManager
+from .models import LogStream, WorkspaceLogManager
 from .schemas import LogEntry, LogProblem, LogStage
 
 __all__ = [
+    "LogStream",
     "WorkspaceLogManager",
     "LogEntry",
     "LogProblem",