py: add wait param to PipelineBuilder.create

abhizer · gz · commit d6401a2d3795 · 2025-11-04T21:56:40.000Z
Adds the parameter `wait` to PipelineBuilder.create, create_or_replace methods. `wait` determines if the function should way for the pipeline to compile after being created. True by default. Fixes: feldera#5018 Also updates the build-docs workflow to only build the feldera package instead of the entire feldera/python directory. Fixes: feldera#5019 Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com>
diff --git a/.github/workflows/build-docs.yml b/.github/workflows/build-docs.yml
@@ -20,7 +20,7 @@ jobs:
       - name: Build Python SDK docs
         working-directory: ./python/docs
         run: |
-          uv run sphinx-apidoc -o . ../
+          uv run sphinx-apidoc -o . ../feldera
           uv run sphinx-build -M html . _build
 
       - name: Copy the Python docs to static
diff --git a/python/feldera/pipeline_builder.py b/python/feldera/pipeline_builder.py
@@ -1,12 +1,12 @@
 import os
 from typing import Optional
 
+from feldera.enums import CompilationProfile, PipelineFieldSelector
+from feldera.pipeline import Pipeline
+from feldera.rest.errors import FelderaAPIError
 from feldera.rest.feldera_client import FelderaClient
 from feldera.rest.pipeline import Pipeline as InnerPipeline
-from feldera.pipeline import Pipeline
-from feldera.enums import CompilationProfile, PipelineFieldSelector
 from feldera.runtime_config import RuntimeConfig
-from feldera.rest.errors import FelderaAPIError
 
 
 class PipelineBuilder:
@@ -49,10 +49,11 @@ def __init__(
             "FELDERA_RUNTIME_VERSION", runtime_version
         )
 
-    def create(self) -> Pipeline:
+    def create(self, wait: bool = True) -> Pipeline:
         """
         Create the pipeline if it does not exist.
 
+        :param wait: Whether to wait for the pipeline to be created. True by default
         :return: The created pipeline
         """
 
@@ -82,17 +83,20 @@ def create(self) -> Pipeline:
             runtime_config=self.runtime_config.to_dict(),
         )
 
-        inner = self.client.create_pipeline(inner)
+        inner = self.client.create_pipeline(inner, wait=wait)
         pipeline = Pipeline(self.client)
         pipeline._inner = inner
 
         return pipeline
 
-    def create_or_replace(self) -> Pipeline:
+    def create_or_replace(self, wait: bool = True) -> Pipeline:
         """
         Creates a pipeline if it does not exist and replaces it if it exists.
 
         If the pipeline exists and is running, it will be stopped and replaced.
+
+        :param wait: Whether to wait for the pipeline to be created. True by default
+        :return: The created pipeline
         """
 
         if self.name is None or self.sql is None:
@@ -121,7 +125,7 @@ def create_or_replace(self) -> Pipeline:
             runtime_config=self.runtime_config.to_dict(),
         )
 
-        inner = self.client.create_or_update_pipeline(inner)
+        inner = self.client.create_or_update_pipeline(inner, wait=wait)
         pipeline = Pipeline(self.client)
         pipeline._inner = inner
 
diff --git a/python/feldera/rest/feldera_client.py b/python/feldera/rest/feldera_client.py
@@ -1,20 +1,20 @@
-import pathlib
-from typing import Any, Dict, Optional
+import json
 import logging
+import pathlib
 import time
-import json
 from decimal import Decimal
-from typing import Generator, Mapping
+from typing import Any, Dict, Generator, Mapping, Optional
 from urllib.parse import quote
+
 import requests
 
 from feldera.enums import BootstrapPolicy, PipelineFieldSelector, PipelineStatus
+from feldera.rest._helpers import client_version
+from feldera.rest._httprequests import HttpRequests
 from feldera.rest.config import Config
+from feldera.rest.errors import FelderaAPIError, FelderaTimeoutError
 from feldera.rest.feldera_config import FelderaConfig
-from feldera.rest.errors import FelderaTimeoutError, FelderaAPIError
 from feldera.rest.pipeline import Pipeline
-from feldera.rest._httprequests import HttpRequests
-from feldera.rest._helpers import client_version
 
 
 def _validate_no_none_keys_in_map(data):
@@ -258,12 +258,12 @@ def __wait_for_pipeline_state_one_of(
             )
             time.sleep(0.1)
 
-    def create_pipeline(self, pipeline: Pipeline) -> Pipeline:
+    def create_pipeline(self, pipeline: Pipeline, wait: bool = True) -> Pipeline:
         """
         Create a pipeline if it doesn't exist and wait for it to compile
 
-
-        :name: The name of the pipeline
+        :param pipeline: The pipeline to create
+        :param wait: Whether to wait for the pipeline to compile. True by default
         """
 
         body = {
@@ -281,12 +281,21 @@ def create_pipeline(self, pipeline: Pipeline) -> Pipeline:
             body=body,
         )
 
+        if not wait:
+            return pipeline
+
         return self.__wait_for_compilation(pipeline.name)
 
-    def create_or_update_pipeline(self, pipeline: Pipeline) -> Pipeline:
+    def create_or_update_pipeline(
+        self, pipeline: Pipeline, wait: bool = True
+    ) -> Pipeline:
         """
         Create a pipeline if it doesn't exist or update a pipeline and wait for
         it to compile
+
+        :param pipeline: The pipeline to create or update
+        :param wait: Whether to wait for the pipeline to compile. True by default
+        :return: The created or updated pipeline
         """
 
         body = {
@@ -304,6 +313,9 @@ def create_or_update_pipeline(self, pipeline: Pipeline) -> Pipeline:
             body=body,
         )
 
+        if not wait:
+            return pipeline
+
         return self.__wait_for_compilation(pipeline.name)
 
     def patch_pipeline(
