feat(pipeline-stream): add log stream

Author: Datata1

examples/create_workspace_with_landscape.py

@@ -1,61 +1,98 @@
+"""
+Demo: Create a workspace, deploy a landscape profile, and stream logs.
+"""
+
 import asyncio
 import time
 
 from codesphere import CodesphereSDK
 from codesphere.resources.workspace import WorkspaceCreate
-from codesphere.resources.workspace.landscape import ProfileBuilder, ProfileConfig
+from codesphere.resources.workspace.landscape import (
+    PipelineStage,
+    PipelineState,
+    ProfileBuilder,
+)
+from codesphere.resources.workspace.logs import LogStage
 
-TEAM_ID = 123  # Replace with your actual team ID
+TEAM_ID = 35698
 
 
-async def get_plan_id(sdk: CodesphereSDK, plan_name: str = "Micro") -> int:
-    plans = await sdk.metadata.list_plans()
-    plan = next((p for p in plans if p.title == plan_name and not p.deprecated), None)
-    if not plan:
-        raise ValueError(f"Plan '{plan_name}' not found")
-    return plan.id
+async def main():
+    async with CodesphereSDK() as sdk:
+        plans = await sdk.metadata.list_plans()
+        plan = next((p for p in plans if p.title == "Micro" and not p.deprecated), None)
+        if not plan:
+            raise ValueError("Micro plan not found")
 
+        workspace_name = f"pipeline-demo-{int(time.time())}"
+        print(f"Creating workspace '{workspace_name}'...")
 
-def build_web_profile(plan_id: int) -> ProfileConfig:
-    """Build a simple web service landscape profile."""
-    return (
-        ProfileBuilder()
-        .prepare()
-        .add_step("npm install", name="Install dependencies")
-        .done()
-        .add_reactive_service("web")
-        .plan(plan_id)
-        .add_step("npm start")
-        .add_port(3000, public=True)
-        .add_path("/", port=3000)
-        .replicas(1)
-        .env("NODE_ENV", "production")
-        .build()
-    )
+        workspace = await sdk.workspaces.create(
+            WorkspaceCreate(plan_id=plan.id, team_id=TEAM_ID, name=workspace_name)
+        )
+        print(f"✓ Workspace created (ID: {workspace.id})")
 
+        print("Waiting for workspace to start...")
+        await workspace.wait_until_running(timeout=300.0, poll_interval=5.0)
+        print("✓ Workspace is running\n")
 
-async def create_workspace(sdk: CodesphereSDK, plan_id: int, name: str):
-    workspace = await sdk.workspaces.create(
-        WorkspaceCreate(plan_id=plan_id, team_id=TEAM_ID, name=name)
-    )
-    await workspace.wait_until_running(timeout=300.0, poll_interval=5.0)
-    return workspace
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("echo 'Installing dependencies...' && sleep 2")
+            .add_step("echo 'Setup complete!' && sleep 1")
+            .done()
+            .add_reactive_service("web")
+            .plan(plan.id)
+            .add_step(
+                'for i in $(seq 1 50); do echo "[$i] Processing request..."; sleep 1; done'
+            )
+            .add_port(3000, public=True)
+            .add_path("/", port=3000)
+            .replicas(1)
+            .done()
+            .build()
+        )
 
+        print("Deploying landscape profile...")
+        await workspace.landscape.save_profile("production", profile)
+        await workspace.landscape.deploy(profile="production")
+        print("✓ Profile deployed\n")
 
-async def deploy_landscape(workspace, profile: dict, profile_name: str = "production"):
-    await workspace.landscape.save_profile(profile_name, profile)
-    await workspace.landscape.deploy(profile=profile_name)
-    print("Deployment started!")
+        print("--- Prepare Stage ---")
+        await workspace.landscape.start_stage(
+            PipelineStage.PREPARE, profile="production"
+        )
+        prepare_status = await workspace.landscape.wait_for_stage(
+            PipelineStage.PREPARE, timeout=60.0
+        )
 
+        for status in prepare_status:
+            icon = "✓" if status.state == PipelineState.SUCCESS else "✗"
+            print(f"{icon} {status.server}: {status.state.value}")
 
-async def main():
-    async with CodesphereSDK() as sdk:
-        plan_id = await get_plan_id(sdk)
-        workspace = await create_workspace(
-            sdk, plan_id, f"landscape-demo-{int(time.time())}"
-        )
-        profile = build_web_profile(plan_id)
-        await deploy_landscape(workspace, profile)
+        print("\nPrepare logs:")
+        for step in range(len(prepare_status[0].steps)):
+            logs = await workspace.logs.collect(
+                stage=LogStage.PREPARE, step=step, timeout=5.0
+            )
+            for entry in logs:
+                if entry.get_text():
+                    print(f"  {entry.get_text().strip()}")
+
+        print("\n--- Run Stage ---")
+        await workspace.landscape.start_stage(PipelineStage.RUN, profile="production")
+        print("Started run stage, waiting for logs...\n")
+
+        print("Streaming logs from 'web' service:")
+        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
+
+        print(f"\n✓ Stream ended ({count} log entries)")
+        print(f"✓ Workspace {workspace.id} is still running.")
 
 
 if __name__ == "__main__":

src/codesphere/resources/workspace/__init__.py

@@ -1,4 +1,5 @@
 from .git import GitHead, WorkspaceGitManager
+from .logs import LogEntry, LogProblem, LogStage, WorkspaceLogManager
 from .resources import WorkspacesResource
 from .schemas import (
     CommandInput,
@@ -19,4 +20,8 @@
     "CommandOutput",
     "WorkspaceGitManager",
     "GitHead",
+    "WorkspaceLogManager",
+    "LogEntry",
+    "LogProblem",
+    "LogStage",
 ]

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

@@ -4,6 +4,10 @@
     ManagedServiceConfig,
     NetworkConfig,
     PathConfig,
+    PipelineStage,
+    PipelineState,
+    PipelineStatus,
+    PipelineStatusList,
     PortConfig,
     Profile,
     ProfileBuilder,
@@ -12,6 +16,7 @@
     ReactiveServiceConfig,
     StageConfig,
     Step,
+    StepStatus,
 )
 
 __all__ = [
@@ -28,4 +33,9 @@
     "NetworkConfig",
     "PortConfig",
     "PathConfig",
+    "PipelineStage",
+    "PipelineState",
+    "PipelineStatus",
+    "PipelineStatusList",
+    "StepStatus",
 ]

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

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 import re
 from typing import TYPE_CHECKING, Dict, List, Optional, Union
@@ -10,10 +11,20 @@
 from .operations import (
     _DEPLOY_OP,
     _DEPLOY_WITH_PROFILE_OP,
+    _GET_PIPELINE_STATUS_OP,
     _SCALE_OP,
+    _START_PIPELINE_STAGE_OP,
+    _START_PIPELINE_STAGE_WITH_PROFILE_OP,
+    _STOP_PIPELINE_STAGE_OP,
     _TEARDOWN_OP,
 )
-from .schemas import Profile, ProfileConfig
+from .schemas import (
+    PipelineStage,
+    PipelineState,
+    PipelineStatusList,
+    Profile,
+    ProfileConfig,
+)
 
 if TYPE_CHECKING:
     from ..schemas import CommandOutput
@@ -95,3 +106,155 @@ 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
+
+        if profile is not None:
+            _validate_profile_name(profile)
+            await self._execute_operation(
+                _START_PIPELINE_STAGE_WITH_PROFILE_OP, stage=stage, profile=profile
+            )
+        else:
+            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
+
+        await self._execute_operation(_STOP_PIPELINE_STAGE_OP, stage=stage)
+
+    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
+
+        return await self._execute_operation(_GET_PIPELINE_STATUS_OP, stage=stage)
+
+    async def wait_for_stage(
+        self,
+        stage: Union[PipelineStage, str],
+        *,
+        timeout: float = 300.0,
+        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")
+
+        stage_name = stage.value if isinstance(stage, PipelineStage) else stage
+        elapsed = 0.0
+
+        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...",
+                    stage_name,
+                )
+                await asyncio.sleep(poll_interval)
+                elapsed += poll_interval
+                continue
+
+            # Check if all relevant servers have completed
+            all_completed = all(
+                s.state
+                in (PipelineState.SUCCESS, PipelineState.FAILURE, PipelineState.ABORTED)
+                for s in relevant_statuses
+            )
+
+            if all_completed:
+                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)",
+                stage_name,
+                ", ".join(states),
+                elapsed,
+            )
+            await asyncio.sleep(poll_interval)
+            elapsed += poll_interval
+
+        raise TimeoutError(
+            f"Pipeline stage '{stage_name}' did not complete within {timeout} seconds."
+        )