feldera.pipeline_builder module

class feldera.pipeline_builder.PipelineBuilder(client: ~feldera.rest.feldera_client.FelderaClient, name: str, sql: str, udf_rust: str = '', udf_toml: str = '', description: str = '', tags: ~typing.List[str] | None = None, compilation_profile: ~feldera.enums.CompilationProfile = CompilationProfile.OPTIMIZED, runtime_config: ~feldera.runtime_config.RuntimeConfig = <feldera.runtime_config.RuntimeConfig object>, runtime_version: str | None = None)[source]

Bases: object

A builder for creating a Feldera Pipeline.

Parameters:

• client – The FelderaClient instance

• name – The name of the pipeline

• description – The description of the pipeline

• tags – Free-form labels used to organize, group, and filter pipelines

• sql – The SQL code of the pipeline

• udf_rust – Rust code for UDFs

• udf_toml – Rust dependencies required by UDFs (in the TOML format)

• compilation_profile – The CompilationProfile to use

• runtime_config – The RuntimeConfig to use. Enables configuring the runtime behavior of the pipeline such as: fault tolerance, storage and Resources

__init__(client: ~feldera.rest.feldera_client.FelderaClient, name: str, sql: str, udf_rust: str = '', udf_toml: str = '', description: str = '', tags: ~typing.List[str] | None = None, compilation_profile: ~feldera.enums.CompilationProfile = CompilationProfile.OPTIMIZED, runtime_config: ~feldera.runtime_config.RuntimeConfig = <feldera.runtime_config.RuntimeConfig object>, runtime_version: str | None = None)[source]

create(wait: bool = True) → Pipeline[source]

Create the pipeline if it does not exist.

Parameters:

wait – Whether to wait for the pipeline to be created. True by default

Returns:

The created pipeline

create_or_replace(wait: bool = True) → Pipeline[source]

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.

Parameters:

wait – Whether to wait for the pipeline to be created. True by default

Returns:

The created pipeline

feldera.pipeline module

class feldera.pipeline.Pipeline(client: FelderaClient)[source]

Bases: object

__init__(client: FelderaClient)[source]

activate(wait: bool = True, timeout_s: float | None = None) → PipelineStatus | None[source]

Activates the pipeline when starting from STANDBY mode. Only applicable when the pipeline is starting from a checkpoint in object store.

Parameters:

• wait – Set True to wait for the pipeline to activate. True by default

• timeout_s – The maximum time (in seconds) to wait for the pipeline to pause.

advance_clock(delta_ms: int | None = None) → Dict[str, int | str][source]

Advance the externally-driven NOW() clock and return its new value.

Requires dev_tweaks.now_http_driven = True on this pipeline. Forward-only: delta_ms must be non-negative.

Parameters:

delta_ms – Milliseconds to add to NOW(). 0 reads the current value without moving the clock; None (the default) advances by one clock_resolution (one configured tick). Any non-zero value rounds up to the next clock_resolution boundary, so a sub-resolution delta still moves the clock by one full tick.

The returned now_ms is the value the worker will emit on its next pipeline step; queries against materialized views may observe the previous NOW() until that step completes. Poll the view if you need read-after-write semantics.

Returns:

A dict {"now_ms": <int>, "now": <RFC 3339 str>}.

Raises:

FelderaAPIError – If the clock is not in http-driven mode, the body is malformed, or the pipeline is not running.

static all(client: FelderaClient) → List[Pipeline][source]

Get all pipelines.

Parameters:

client – The FelderaClient instance.

Returns:

A list of Pipeline objects.

approve(silent_bootstrap: bool = False, concurrent_bootstrap: bool = False)[source]

Approves the pipeline to proceed with bootstrapping.

This method is used when a pipeline has been started with bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL and is currently in the AWAITINGAPPROVAL state. The pipeline will wait for explicit user approval before proceeding with the bootstrapping process.

Parameters:

• silent_bootstrap – Set True to bootstrap with output connectors disabled, so no records are emitted during the bootstrap phase. False by default.

• concurrent_bootstrap – Set True to bootstrap new and modified views concurrently, keeping the pre-existing views live while the new ones backfill in the background. Mutually exclusive with silent_bootstrap. False by default.

bootstrap_policy() → BootstrapPolicy | None[source]

Return the bootstrap policy of the pipeline.

checkpoint(wait: bool = False, timeout_s: float | None = None) → int[source]

Checkpoints this pipeline, returning just the checkpoint sequence number.

This is for backward compatibility: checkpoint_response is a better choice for a caller that plans to check the checkpoint’s status itself later (e.g. after this call returns with wait=False), since it also returns the incarnation_uuid needed for that later check to detect a pipeline restart in between.

Parameters:

• wait – If true, will block until the checkpoint completes.

• timeout_s – The maximum time (in seconds) to wait for the checkpoint to complete (defaults to None = no timeout is enforced).

Returns:

The checkpoint sequence number.

Raises:

• FelderaAPIError – If enterprise features are not enabled.

• RuntimeError – If wait is true and the checkpoint fails.

checkpoint_response(wait: bool = False, timeout_s: float | None = None) → dict[source]

Checkpoints this pipeline, returning the full response.

Parameters:

• wait – If true, will block until the checkpoint completes.

• timeout_s – The maximum time (in seconds) to wait for the checkpoint to complete (defaults to None = no timeout is enforced).

Returns:

dict with checkpoint_sequence_number and incarnation_uuid; pass the latter to checkpoint_status to detect a pipeline restart between this call and a later status check.

Raises:

• FelderaAPIError – If enterprise features are not enabled.

• RuntimeError – If wait is true and the checkpoint fails.

checkpoint_status(seq: int, incarnation_uuid: str | None = None) → CheckpointStatus[source]

Checks the status of the given checkpoint.

Parameters:

• seq – The checkpoint sequence number.

• incarnation_uuid – The incarnation_uuid returned alongside seq by checkpoint, if known, to allow pipeline restarts to be clearly reported as IncarnationUuidMismatch.

checkpoints() → List[CheckpointMetadata][source]

Returns the list of checkpoints for this pipeline.

clear_storage(wait: bool = True, timeout_s: float | None = None, poll_interval_s: float = 0.25)[source]

Clears the storage of the pipeline. Once started, this action cannot be canceled, and will delete all the pipeline storage.

Parameters:

• wait – Set True to wait for the pipeline storage to become cleared, or set False to immediately return. Default is True.

• timeout_s – Timeout waiting for storage to become cleared. None = no timeout is enforced (default). Not used if wait=False.

• poll_interval_s – Polling interval at which to check while waiting if storage is cleared (default is every 0.25 seconds). Not used if wait=False.

commit_transaction(transaction_id: int | None = None, wait: bool = True, timeout_s: float | None = None, poll_interval_s: float = 0.5, incarnation_uuid: str | None = None)[source]

Commit the currently active transaction.

Parameters:

• transaction_id – If provided, the function verifies that the currently active transaction matches this ID. If the active transaction ID does not match, the function raises an error.

• wait – If True, the function blocks until the transaction either commits successfully or the timeout is reached. If False, the function initiates the commit and returns immediately without waiting for completion. The default value is True.

• timeout_s – Maximum time (in seconds) to wait for the transaction to commit when wait is True. If None, the function will wait indefinitely.

• poll_interval_s – Polling interval at which to check while waiting for the transaction to commit (default is every 0.5 seconds). Not used if wait=False.

• incarnation_uuid – The incarnation_uuid that start_transaction_response returned. If the pipeline restarted since then, the commit fails with IncarnationUuidMismatch. If not provided, the function uses the pipeline’s incarnation when the call starts.

Raises:

• RuntimeError – If transaction_id is not provided and there is currently no transaction in progress, or if the pipeline restarts while the function waits for the commit.

• ValueError – If the provided transaction_id does not match the current transaction. If there is no current transaction, the pipeline checks transaction_id: if the last commit request already committed it, the commit succeeds, so it is safe to call this again after a TimeoutError.

• TimeoutError – If the transaction does not commit within the specified timeout (when wait is True).

• FelderaAPIError – If the pipeline fails to commit a transaction.

completion_token_status(token: str) → CompletionTokenStatus[source]

Returns the status of the completion token.

created_at() → datetime[source]

Return the creation time of the pipeline.

delete(clear_storage: bool = False)[source]

Deletes the pipeline.

The pipeline must be stopped, and the storage cleared before it can be deleted.

Parameters:

clear_storage – True if the storage should be cleared before deletion. False by default

Raises:

FelderaAPIError – If the pipeline is not in STOPPED state or the storage is still bound.

deployment_config() → Mapping[str, Any][source]

Return the deployment config of the pipeline.

deployment_desired_status() → DeploymentDesiredStatus[source]

Return the desired deployment status of the pipeline. This is the next state that the pipeline should transition to.

deployment_error() → Mapping[str, Any][source]

Return the deployment error of the pipeline. Returns an empty string if there is no error.

deployment_location() → str[source]

Return the deployment location of the pipeline. Deployment location is the location where the pipeline can be reached at runtime (a TCP port number or a URI).

deployment_resources_desired_status() → DeploymentResourcesDesiredStatus[source]

Return the desired status of the the deployment resources.

deployment_resources_status() → DeploymentResourcesStatus[source]

Return the status of the deployment resources.

deployment_runtime_desired_status() → DeploymentRuntimeDesiredStatus[source]

Return the deployment runtime desired status.

deployment_runtime_status() → DeploymentRuntimeStatus[source]

Return the deployment runtime status.

deployment_runtime_status_details() → dict | None[source]

Return the deployment runtime status details.

deployment_status_since() → datetime[source]

Return the timestamp when the current deployment status of the pipeline was set.

description() → str[source]

Return the description of the pipeline.

diff(program_code: str | None = None, runtime_version: str | None = None) → Dict[str, Any][source]

Compute the diff between this pipeline’s current program and a proposed new version, without modifying or restarting the pipeline.

The diff lists the tables, views, and connectors that would be added, removed, or modified. It is the same diff shown when approving changes during bootstrapping, so it lets you preview the effect of a change before applying it.

The baseline is this pipeline’s currently configured program compiled with its runtime, not necessarily the program in the latest checkpoint (which may have been produced by a different program or runtime).

Parameters:

• program_code – New SQL program code to compare against. If None (the default), the pipeline’s current program code is used.

• runtime_version – Runtime version to compile the new program with: a version tag (vX.Y.Z) or a 40-character git SHA. If None (the default), the platform’s default runtime is used.

Returns:

The pipeline diff as a dict (see the PipelineDiff schema).

Raises:

FelderaAPIError – If the current program is not compiled, the new program fails to compile, the change cannot be bootstrapped, or the compiler service is unavailable.

dismiss_error()[source]

Dismisses the deployment_error of the pipeline.

errors() → List[Mapping[str, Any]][source]

Returns a list of all errors in this pipeline.

event(event_id: str, *, selector: str = 'status') → dict[source]

Retrieves a specific pipeline event.

Parameters:

• event_id – Identifier (UUID) of the event to retrieve, or latest for the latest event.

• selector – (Optional) Limit the returned fields. Valid values: “all”, “status” (default).

Returns:

Event (fields limited based on selector).

events(*, selector: str = 'status') → List[dict][source]

Retrieves all pipeline events (status fields only).

Parameters:

selector – (Optional) Limit the returned fields. Valid values: “all”, “status” (default).

Returns:

List of pipeline events.

execute(query: str, wait: bool = False)[source]

Executes an ad-hoc SQL query on the current pipeline, discarding its result. Unlike the query() method which returns a generator for retrieving query results lazily, this method processes the query eagerly and fully before returning.

This method is suitable for SQL operations like INSERT where the user needs confirmation of successful query execution, but does not require the query result. If the query fails, an exception will be raised.

Important:

If you try to INSERT data into a table while the pipeline is paused, it will block until the pipeline is resumed. Ad-hoc queries cannot delete data.

Parameters:

• query – The SQL query to be executed.

• wait – When True, block until the query’s writes have been fully processed by the pipeline. Once the call returns, a subsequent ad-hoc query observes the writes (read-your-writes) and every output connector has processed the outputs the writes produced. False by default for backward compatibility.

Raises:

• FelderaAPIError – If the pipeline is not in a RUNNING state.

• FelderaAPIError – If the query is invalid.

foreach_chunk(view_name: str, callback: Callable[[DataFrame, int], None])[source]

Run the given callback on each chunk of the output of the specified view.

The callback will only receive changes from the point in time when the listener is created. In order to receive all changes since the pipeline started, you can create the pipeline in the PAUSED state using start_paused(), attach listeners and unpause the pipeline using resume().

Parameters:

• view_name – The name of the view.

• callback –

  The callback to run on each chunk. The callback should take two arguments:

  • chunk -> The chunk as a pandas DataFrame

  • seq_no -> The sequence number. The sequence number is a monotonically increasing integer that starts from 0. Note that the sequence number is unique for each chunk, but not necessarily contiguous.

Please note that the callback is run in a separate thread, so it should be thread-safe. Please note that the callback should not block for a long time, as by default, backpressure is enabled and will block the pipeline.

Note

• The callback must be thread-safe as it will be run in a separate thread.

generate_completion_token(table_name: str, connector_name: str) → str[source]

Returns a completion token that can be passed to Pipeline.completion_token_status() to check whether the pipeline has finished processing all inputs received from the connector before the token was generated.

static get(name: str, client: FelderaClient) → Pipeline[source]

Get the pipeline if it exists.

Parameters:

• name – The name of the pipeline.

• client – The FelderaClient instance.

get_samply_profile(output_path: str | None = None) → bytes[source]

Returns the gzip file of the samply profile as bytes.

The gzip file contains the samply profile that can be inspected by the samply tool.

Parameters:

output_path – Optional path to save the samply profile file. If None, the samply profile is only returned as bytes.

Returns:

The samply profile as bytes (GZIP file)

Raises:

FelderaAPIError – If the pipeline does not exist or if there’s an error

id() → str[source]

Return the ID of the pipeline.

input_connector_stats(table_name: str, connector_name: str) → InputEndpointStatus[source]

Get the status of the specified input connector.

input_json(table_name: str, data: Dict | list, update_format: str = 'raw', force: bool = False, wait: bool = True, connector_metadata: Mapping[str, Any] | None = None, **kwargs) → str[source]

Push this JSON data to the specified table of the pipeline.

The pipeline must either be in RUNNING or PAUSED states to push data. An error will be raised if the pipeline is in any other state.

Parameters:

• table_name – The name of the table to push data into.

• data – The JSON encoded data to be pushed to the pipeline. The data should be in the form: {‘col1’: ‘val1’, ‘col2’: ‘val2’} or [{‘col1’: ‘val1’, ‘col2’: ‘val2’}, {‘col1’: ‘val1’, ‘col2’: ‘val2’}]

• update_format – The update format of the JSON data to be pushed to the pipeline. Must be one of: “raw”, “insert_delete”. https://docs.feldera.com/formats/json#the-insertdelete-format

• force – True to push data even if the pipeline is paused. False by default.

• wait – If True, blocks until this input has been processed by the pipeline

• connector_metadata – Connector metadata for every record of this request, as a dict of attribute names to values. CONNECTOR_METADATA() returns it for the records, so a column declared with DEFAULT CAST(CONNECTOR_METADATA()['name'] AS type) takes its value from the name attribute. None by default.

• kwargs – Additional key word arguments forwarded to the Client push_to_pipeline method

Returns:

The completion token to this input.

Raises:

• ValueError – If the update format is invalid.

• FelderaAPIError – If the pipeline is not in a valid state to push data.

• RuntimeError – If the pipeline is paused and force is not set to True.

input_pandas(table_name: str, df: DataFrame, force: bool = False, connector_metadata: Mapping[str, Any] | None = None) → str | None[source]

Push all rows in a pandas DataFrame to the pipeline.

The pipeline must either be in RUNNING or PAUSED states to push data. An error will be raised if the pipeline is in any other state.

The dataframe must have the same columns as the table in the pipeline.

Parameters:

• table_name – The name of the table to insert data into.

• df – The pandas DataFrame to be pushed to the pipeline.

• force – True to push data even if the pipeline is paused. False by default.

• connector_metadata – Connector metadata for every row of the DataFrame, as a dict of attribute names to values. CONNECTOR_METADATA() returns it for the rows. None by default.

Returns:

The completion token of the last chunk pushed, which also covers the earlier chunks, or None if the DataFrame is empty and nothing was pushed.

Raises:

• ValueError – If the table does not exist in the pipeline.

• RuntimeError – If the pipeline is not in a valid state to push data.

• RuntimeError – If the pipeline is paused and force is not set to True.

is_complete() → bool[source]

Check if the pipeline has completed processing all input records.

Returns True if (1) all input connectors attached to the pipeline have finished reading their input data sources and issued end-of-input notifications to the pipeline, and (2) all inputs received from these connectors have been fully processed and corresponding outputs have been sent out through the output connectors.

last_successful_checkpoint_sync() → UUID[source]

Returns the UUID of the last successfully synced checkpoint.

Returns:

The UUID of the last successfully synced checkpoint.

listen(view_name: str) → OutputHandler[source]

Follow the change stream (i.e., the output) of the provided view. Returns an output handle to read the changes.

When the pipeline is stopped, the handle is dropped.

The handle will only receive changes from the point in time when the listener is created. In order to receive all changes since the pipeline started, you can create the pipeline in the PAUSED state using start_paused(), attach listeners and unpause the pipeline using resume().

To read the changes that one input caused, pass the input’s completion token to OutputHandler.wait_for_token() before reading the handle.

Parameters:

view_name – The name of the view to listen to.

logs(follow: bool = True) → Generator[str, None, None][source]

Gets the pipeline logs.

Parameters:

follow – Set False to stop after the retained logs instead of waiting for new lines.

metrics(format: str = 'prometheus')[source]

Gets the pipeline circuit metrics.

Parameters:

format – The metrics format, “prometheus” (default) or “json”.

Returns:

Prometheus exposition text as str, or a dict for “json”.

modify(sql: str | None = None, udf_rust: str | None = None, udf_toml: str | None = None, program_config: Mapping[str, Any] | None = None, runtime_config: Mapping[str, Any] | None = None, description: str | None = None, tags: List[str] | None = None)[source]

Modify the pipeline.

Modify the values of pipeline attributes: SQL code, UDF Rust code, UDF Rust dependencies (TOML), program config, runtime config, description, and tags. Only the provided attributes will be modified. Other attributes will remain unchanged.

Most attributes require the pipeline to be in the STOPPED state to be modified. The description and tags are an exception: they can be modified at any time, as they have no effect on the deployed pipeline.

Raises:

FelderaAPIError – If a non-metadata attribute is modified while the pipeline is not in a STOPPED state.

property name: str

Return the name of the pipeline.

output_connector_stats(view_name: str, connector_name: str) → OutputEndpointStatus[source]

Get the status of the specified output connector.

pause(wait: bool = True, timeout_s: float | None = None)[source]

Pause the pipeline.

The pipeline can only transition to the PAUSED state from the RUNNING state. If the pipeline is already paused, it will remain in the PAUSED state.

Parameters:

• wait – Set True to wait for the pipeline to pause. True by default

• timeout_s – The maximum time (in seconds) to wait for the pipeline to pause.

pause_connector(table_name: str, connector_name: str)[source]

Pause the specified input connector.

Deprecated since version Use: pause_input_connector(), which names the kind of connector it acts on, like pause_output_connector().

pause_input_connector(table_name: str, connector_name: str)[source]

Pause the specified input connector.

A paused input connector stops fetching data from its source. All connectors are RUNNING by default.

Refer to the connector documentation for more information:

https://docs.feldera.com/connectors/orchestration#input-connectors

Parameters:

• table_name – The name of the table that the connector is attached to.

• connector_name – The name of the connector to pause.

Raises:

FelderaAPIError – If the connector is not found, or if the pipeline is not running.

pause_output_connector(view_name: str, connector_name: str)[source]

Pause the specified output connector.

A paused output connector discards the output it receives instead of writing it to its sink. Output produced while the connector is paused is gone for good. All connectors are RUNNING by default.

Refer to the connector documentation for more information:

https://docs.feldera.com/connectors/orchestration#output-connectors

Parameters:

• view_name – The name of the view that the connector is attached to.

• connector_name – The name of the connector to pause.

Raises:

FelderaAPIError – If the connector is not found, or if the pipeline is not running.

platform_version() → str[source]

Return the Feldera platform witch which the program was compiled.

program_code() → str[source]

Return the program SQL code of the pipeline.

program_config() → Mapping[str, Any][source]

Return the program config of the pipeline.

program_error() → Mapping[str, Any][source]

Return the program error of the pipeline. If there are no errors, the exit_code field inside both sql_compilation and rust_compilation will be 0.

program_info() → Mapping[str, Any][source]

Return the program info of the pipeline. This is the output returned by the SQL compiler, including: the list of input and output connectors, the generated Rust code for the pipeline, and the SQL program schema.

program_status() → ProgramStatus[source]

Return the program status of the pipeline.

Program status is the status of compilation of this SQL program. We first compile the SQL program to Rust code, and then compile the Rust code to a binary.

program_status_since() → datetime[source]

Return the timestamp when the current program status was set.

program_version() → int[source]

Return the program version of the pipeline.

query(query: str, wait: bool = False) → Generator[Mapping[str, Any], None, None][source]

Executes an ad-hoc SQL query on this pipeline and returns a generator that yields the rows of the result as Python dictionaries. For INSERT queries, consider using execute() instead. All floating-point numbers are deserialized as Decimal to avoid precision loss.

For new code, prefer query_arrow(): Arrow IPC keeps full SQL type fidelity, lets MAP keys be non-string values, and returns every column even when two share a name. See https://github.com/feldera/feldera/issues/4219.

Note:

You can only SELECT from materialized tables and views.

Important:

With the default wait=False this method is lazy: it returns a generator and is not evaluated until you consume the result.

Parameters:

• query – The SQL query to be executed.

• wait – When True, block until the query’s writes have been fully processed by the pipeline. Once the call returns, a subsequent ad-hoc query observes the writes (read-your-writes) and every output connector has processed the outputs the writes produced. This is useful for INSERT queries; it has no effect on a pure SELECT other than waiting for already-in-flight records to be processed. Setting wait=True consumes the result eagerly. False by default for backward compatibility.

Returns:

A generator that yields the rows of the result as Python dictionaries.

Raises:

• FelderaAPIError – If the pipeline is not in a RUNNING or PAUSED state.

• FelderaAPIError – If querying a non materialized table or view.

• FelderaAPIError – If the query is invalid.

query_arrow(query: str) → Generator[RecordBatch, None, None][source]

Executes an ad-hoc SQL query on this pipeline and returns a generator that yields the result as PyArrow RecordBatches.

Note:

You can only SELECT from materialized tables and views.

Parameters:

query – The SQL query to be executed.

Raises:

• FelderaAPIError – If the pipeline is not in a RUNNING or PAUSED state.

• FelderaAPIError – If querying a non materialized table or view.

• FelderaAPIError – If the query is invalid.

Returns:

A generator that yields pyarrow.RecordBatch objects.

query_arrow_dicts(query: str) → Generator[Mapping[str, Any], None, None][source]

Executes an ad-hoc SQL query on this pipeline and returns a generator that yields the result as a sequence of Dictionaries.

Note:

You can only SELECT from materialized tables and views.

Parameters:

query – The SQL query to be executed.

Raises:

• FelderaAPIError – If the pipeline is not in a RUNNING or PAUSED state.

• FelderaAPIError – If querying a non materialized table or view.

• FelderaAPIError – If the query is invalid.

Returns:

A generator that yields pyarrow.Mapping objects.

query_hash(query: str)[source]

Executes an ad-hoc SQL query on this pipeline and returns the result as a hash of the result set. This is useful for quickly checking if the result set has changed without retrieving the entire result.

Note:

For a stable hash, the query must be deterministic which means it should be sorted.

Note:

The hash identifies a result set within one Feldera release. It is built on DataFusion’s row hash, so a Feldera upgrade can change it for unchanged data; compare hashes taken from the same release.

Parameters:

query – The SQL query to be executed.

Raises:

• FelderaAPIError – If the pipeline is not in a RUNNING or PAUSED state.

• FelderaAPIError – If querying a non materialized table or view.

• FelderaAPIError – If the query is invalid.

query_parquet(query: str, path: str)[source]

Executes an ad-hoc SQL query on this pipeline and saves the result to the specified path as a parquet file. If the extension isn’t parquet, it will be automatically appended to path.

Note:

You can only SELECT from materialized tables and views.

Parameters:

• query – The SQL query to be executed.

• path – The path of the parquet file.

Raises:

• FelderaAPIError – If the pipeline is not in a RUNNING or PAUSED state.

• FelderaAPIError – If querying a non materialized table or view.

• FelderaAPIError – If the query is invalid.

query_tabular(query: str) → Generator[str, None, None][source]

Executes a SQL query on this pipeline and returns the result as a formatted string.

Note:

You can only SELECT from materialized tables and views.

Important:

This method is lazy. It returns a generator and is not evaluated until you consume the result.

Parameters:

query – The SQL query to be executed.

Returns:

A generator that yields a string representing the query result in a human-readable, tabular format.

Raises:

• FelderaAPIError – If the pipeline is not in a RUNNING or PAUSED state.

• FelderaAPIError – If querying a non materialized table or view.

• FelderaAPIError – If the query is invalid.

rebalance()[source]

Initiate rebalancing.

Initiate immediate rebalancing of the pipeline. Normally rebalancing is initiated automatically when the drift in the size of joined relations exceeds a threshold. This method forces the balancer to reevaluate and apply an optimal partitioning policy regardless of the threshold.

This operation is a no-op unless the adaptive_joins feature is enabled in dev_tweaks.

refresh(field_selector: PipelineFieldSelector)[source]

Calls the backend to get the updated, latest version of the pipeline.

Parameters:

field_selector – Choose what pipeline information to refresh; see PipelineFieldSelector enum definition.

Raises:

FelderaConnectionError – If there is an issue connecting to the backend.

remote_checkpoints() → list[dict][source]

Lists checkpoints available in the configured remote object storage.

Returns:

A list of remote checkpoint descriptors, each containing at minimum a uuid field.

Raises:

FelderaAPIError – If sync storage is not configured or enterprise features are not enabled.

restart(bootstrap_policy: BootstrapPolicy | None = None, silent_bootstrap: bool = False, concurrent_bootstrap: bool = False, timeout_s: float | None = None, dismiss_error: bool = True, reloadable_tables: List[str] | None = None)[source]

Restarts the pipeline.

This method forcibly STOPS the pipeline regardless of its current state and then starts it again. No checkpoints are made when stopping the pipeline.

Parameters:

• bootstrap_policy – The bootstrap policy to use.

• silent_bootstrap – Set True to bootstrap the pipeline with output connectors disabled. False by default.

• concurrent_bootstrap – Set True to bootstrap new and modified views concurrently, keeping the pre-existing views live while the new ones backfill in the background. Mutually exclusive with silent_bootstrap. False by default.

• timeout_s – The maximum time (in seconds) to wait for the pipeline to restart.

• dismiss_error – Set True to dismiss any deployment error before starting; set False to make it fail in that case. True by default.

• reloadable_tables – Tables that bootstrapping may reload. See start().

resume(wait: bool = True, timeout_s: float | None = None)[source]

Resumes the pipeline from the PAUSED state. If the pipeline is already running, it will remain in the RUNNING state.

Parameters:

• wait – Set True to wait for the pipeline to resume. True by default

• timeout_s – The maximum time (in seconds) to wait for the pipeline to resume.

resume_connector(table_name: str, connector_name: str)[source]

Start (resume) the specified input connector.

Deprecated since version Use: start_input_connector(), which names the kind of connector it acts on, like start_output_connector().

resume_logs(cursor: str | None = None) → LogStream[source]

Open the pipeline logs stream, starting from the last read line (cursor).

Where logs() replays the whole retained buffer on every connection, this delivers only the lines following cursor.

The logs buffer is discarded when the runner restarts or the pipeline is deleted, and survives a pipeline being stopped and started. A cursor issued during an earlier lifetime of the buffer is not an error: the stream starts at the beginning of the retained buffer, and LogPosition.gap names the lines that were lost.

For example, to keep reading the logs across dropped connections:

cursor = None
while True:
    with pipeline.resume_logs(cursor) as stream:
        for line in stream:
            print(line)
        cursor = stream.cursor()

Parameters:

cursor – The position to resume after, as returned by feldera.rest.logs.LogStream.cursor(). None starts at the beginning of the retained buffer.

Returns:

The open stream. Close it when done reading.

runtime_config() → RuntimeConfig[source]

Return the runtime config of the pipeline.

set_runtime_config(runtime_config: RuntimeConfig)[source]

Updates the runtime config of the pipeline. The pipeline must be stopped. Changing some pipeline configuration, such as the number of workers, requires storage to be cleared.

For example, to set ‘min_batch_size_records’ on a pipeline:

runtime_config = pipeline.runtime_config()
runtime_config.min_batch_size_records = 500
pipeline.set_runtime_config(runtime_config)

start(bootstrap_policy: BootstrapPolicy | None = None, silent_bootstrap: bool = False, concurrent_bootstrap: bool = False, wait: bool = True, timeout_s: float | None = None, dismiss_error: bool = True, reloadable_tables: List[str] | None = None)[source]

Starts this pipeline.

• The pipeline must be in STOPPED state to start.

• If the pipeline is in any other state, an error will be raised.

• If the pipeline is in PAUSED state, use .meth:resume instead.

Parameters:

• bootstrap_policy – The bootstrap policy to use.

• silent_bootstrap – Set True to bootstrap the pipeline with output connectors disabled. False by default.

• concurrent_bootstrap – Set True to bootstrap new and modified views concurrently, keeping the pre-existing views live while the new ones backfill in the background. Mutually exclusive with silent_bootstrap. False by default.

• timeout_s – The maximum time (in seconds) to wait for the pipeline to start.

• wait – Set True to wait for the pipeline to start. True by default

• dismiss_error – Set True to dismiss any deployment error before starting; set False to make it fail in that case. True by default.

• reloadable_tables – Tables that bootstrapping may reload. Use it with bootstrap_policy=BootstrapPolicy.ALLOW and wait=True. If a table is modified, bootstrapping will reload the table: it truncates the table and resets its input connectors. Write each name exactly as the pipeline reports it in the approval diff: an unquoted SQL name in lowercase, a quoted name without its quotes. For example, write t1 for CREATE TABLE T1 and MyTable for CREATE TABLE “MyTable”. None by default: ALLOW bootstraps without a check. If no tables are acceptable to reload, set reloadable_tables=[].

Raises:

• RuntimeError – If the pipeline is not in STOPPED state.

• TableReloadNotAllowedError – If a modified table is not in reloadable_tables, or the pipeline cannot compute the program diff. The pipeline stays in the AWAITINGAPPROVAL state: call approve() to bootstrap or stop(force=True) to reject.

• ValueError – If reloadable_tables is set with a bootstrap policy other than ALLOW, or with wait=False.

start_compaction()[source]

Initiate immediate compaction of the pipeline’s state.

start_input_connector(table_name: str, connector_name: str)[source]

Start (resume) the specified input connector.

Refer to the connector documentation for more information:

https://docs.feldera.com/connectors/orchestration#input-connectors

Parameters:

• table_name – The name of the table that the connector is attached to.

• connector_name – The name of the connector to start.

Raises:

FelderaAPIError – If the connector is not found, or if the pipeline is not running.

start_output_connector(view_name: str, connector_name: str)[source]

Start (resume) the specified output connector.

The connector resumes with the output the pipeline produces from this point on; output produced while it was paused is not replayed.

Refer to the connector documentation for more information:

https://docs.feldera.com/connectors/orchestration#output-connectors

Parameters:

• view_name – The name of the view that the connector is attached to.

• connector_name – The name of the connector to start.

Raises:

FelderaAPIError – If the connector is not found, or if the pipeline is not running.

start_paused(bootstrap_policy: BootstrapPolicy | None = None, silent_bootstrap: bool = False, concurrent_bootstrap: bool = False, wait: bool = True, timeout_s: float | None = None, dismiss_error: bool = True, reloadable_tables: List[str] | None = None)[source]

Starts the pipeline in the paused state.

Parameters:

• bootstrap_policy – The bootstrap policy to use.

• silent_bootstrap – Set True to bootstrap the pipeline with output connectors disabled. False by default.

• concurrent_bootstrap – Set True to bootstrap new and modified views concurrently, keeping the pre-existing views live while the new ones backfill in the background. Mutually exclusive with silent_bootstrap. False by default.

• wait – Set True to wait for the pipeline to start. True by default.

• timeout_s – The maximum time (in seconds) to wait for the pipeline to start (defaults to None = no timeout is enforced).

• dismiss_error – Set True to dismiss any deployment error before starting; set False to make it fail in that case. True by default.

• reloadable_tables – Tables that bootstrapping may reload. See start().

start_samply_profile(duration: int = 30)[source]

Starts profiling this pipeline using samply.

Parameters:

duration – The duration of the profile in seconds, from 1 to 300 (default: 30)

Raises:

FelderaAPIError – If the pipeline does not exist or if there’s an error

start_standby(bootstrap_policy: BootstrapPolicy | None = None, silent_bootstrap: bool = False, concurrent_bootstrap: bool = False, wait: bool = True, timeout_s: float | None = None, dismiss_error: bool = True, reloadable_tables: List[str] | None = None)[source]

Starts the pipeline in the standby state.

Parameters:

• bootstrap_policy – The bootstrap policy to use.

• silent_bootstrap – Set True to bootstrap the pipeline with output connectors disabled. False by default.

• concurrent_bootstrap – Set True to bootstrap new and modified views concurrently, keeping the pre-existing views live while the new ones backfill in the background. Mutually exclusive with silent_bootstrap. False by default.

• wait – Set True to wait for the pipeline to start. True by default.

• timeout_s – The maximum time (in seconds) to wait for the pipeline to start (defaults to None = no timeout is enforced).

• dismiss_error – Set True to dismiss any deployment error before starting; set False to make it fail in that case. True by default.

• reloadable_tables – Tables that bootstrapping may reload. See start().

start_transaction() → int[source]

Start a new transaction.

Returns:

Transaction ID.

Raises:

FelderaAPIError – If the pipeline fails to start a transaction, e.g., if the pipeline is not running or there is already an active transaction.

start_transaction_response() → dict[source]

Start a new transaction, returning the full response.

Returns:

dict with transaction_id and incarnation_uuid; pass both to commit_transaction. Transaction IDs start again when the pipeline restarts, so incarnation_uuid prevents a commit of another transaction with the same ID.

Raises:

FelderaAPIError – If the pipeline fails to start a transaction, e.g., if the pipeline is not running or there is already an active transaction.

stats() → PipelineStatistics[source]

Gets the pipeline metrics and performance counters.

status() → PipelineStatus[source]

Return the current status of the pipeline.

stop(force: bool, wait: bool = True, timeout_s: float | None = None)[source]

Stops the pipeline.

Stops the pipeline regardless of its current state.

Parameters:

• force – Set True to immediately scale compute resources to zero. Set False to automatically checkpoint before stopping.

• wait – Set True to gracefully shutdown listeners and wait for the pipeline to stop. True by default.

• timeout_s – The maximum time (in seconds) to wait for the pipeline to stop.

storage_status() → StorageStatus[source]

Return the storage status of the pipeline.

storage_status_details() → dict[source]

Return the storage status details of the pipeline.

support_bundle(output_path: str | None = None, *, collect: bool = True, limit: int | None = None, circuit_profile: bool = True, heap_profile: bool = True, metrics: bool = True, logs: bool = True, stats: bool = True, pipeline_config: bool = True, system_config: bool = True, dataflow_graph: bool = True, pipeline_events: bool = True) → bytes[source]

Generate a support bundle containing diagnostic information from this pipeline.

This method collects various diagnostic data from the pipeline including circuit profile, heap profile, metrics, logs, stats, and connector statistics, and packages them into a single ZIP file for support purposes.

Parameters:

• output_path – Optional path to save the support bundle file. If None, the support bundle is only returned as bytes.

• collect – Whether to collect fresh data from the pipeline (default: True). Set to False to download only previously stored collections.

• limit – If set, include only the N most recent collections.

• circuit_profile – Whether to collect circuit profile data (default: True)

• heap_profile – Whether to collect heap profile data (default: True)

• metrics – Whether to collect metrics data (default: True)

• logs – Whether to collect logs data (default: True)

• stats – Whether to collect stats data (default: True)

• pipeline_config – Whether to collect pipeline configuration data (default: True)

• system_config – Whether to collect system configuration data (default: True)

• dataflow_graph – Whether to collect dataflow graph (default: True)

• pipeline_events – Whether to collect pipeline monitor events (default: True)

Returns:

The support bundle as bytes (ZIP archive)

Raises:

FelderaAPIError – If the pipeline does not exist or if there’s an error

sync_checkpoint(wait: bool = False, timeout_s: float | None = None) → str[source]

Syncs this checkpoint to object store, returning just the checkpoint uuid.

This is for backward compatibility: sync_checkpoint_response is a better choice for a caller that plans to check the sync’s status itself later (e.g. after this call returns with wait=False), since it also returns the incarnation_uuid needed for that later check to detect a pipeline restart in between.

Parameters:

• wait – If true, will block until the checkpoint sync operation completes.

• timeout_s – The maximum time (in seconds) to wait for the checkpoint to complete syncing.

Raises:

• FelderaAPIError – If no checkpoints have been made.

• RuntimeError – If syncing the checkpoint fails.

sync_checkpoint_response(wait: bool = False, timeout_s: float | None = None) → dict[source]

Syncs this checkpoint to object store, returning the full response.

Parameters:

• wait – If true, will block until the checkpoint sync operation completes.

• timeout_s – The maximum time (in seconds) to wait for the checkpoint to complete syncing.

Returns:

dict with checkpoint_uuid and incarnation_uuid; pass the latter to sync_checkpoint_status to detect a pipeline restart between this call and a later status check.

Raises:

• FelderaAPIError – If no checkpoints have been made.

• RuntimeError – If syncing the checkpoint fails.

sync_checkpoint_status(uuid: UUID | str, incarnation_uuid: str | None = None) → CheckpointStatus[source]

Checks the status of the given checkpoint sync operation. If the checkpoint is currently being synchronized, returns CheckpointStatus.InProgress.

CheckpointStatus.Unknown means either that the pipeline is too old to report whether checkpoint sync is in progress, or such a sync never started, or its success or failure indication was displaced by the sync of a different checkpoint.

Failures are not raised as runtime errors and must be explicitly checked.

Parameters:

• uuid – The checkpoint uuid.

• incarnation_uuid – The incarnation_uuid returned alongside uuid by sync_checkpoint, if known, to allow pipeline restarts to be clearly reported as IncarnationUuidMismatch.

tables() → List[SQLTable][source]

Return the tables of the pipeline.

tags() → List[str][source]

Return the tags of the pipeline.

Tags are returned verbatim, including the |rrggbb color suffix the web console uses to encode a tag’s color. Use feldera.tags.tag_display_name() to obtain the tag’s text label.

testing_force_update_platform_version(platform_version: str)[source]

Used to simulate a pipeline compiled with a different platform version than the one currently in use. This is useful for testing platform upgrade behavior without actually upgrading Feldera.

This method is only available when Feldera runs with the “testing” unstable feature enabled.

transaction_id() → int | None[source]

Gets the ID of the transaction that the pipeline runs or commits now, or None if there is no such transaction. This is the transaction that transaction_status describes.

This can be a different transaction than the one that start_transaction started: an earlier transaction, for example one that a connector started, can still be committing. To commit a transaction, use the ID that start_transaction returned.

Returns:

The ID of the transaction.

transaction_status() → TransactionStatus[source]

Get pipeline’s transaction handling status.

Returns:

Current transaction handling status of the pipeline.

Raises:

FelderaAPIError – If pipeline’s status couldn’t be read, e.g., because the pipeline is not currently running.

udf_rust() → str[source]

Return the Rust code for UDFs.

udf_toml() → str[source]

Return the Rust dependencies required by UDFs (in the TOML format).

update_runtime()[source]

Recompile a pipeline with the Feldera runtime version included in the currently installed Feldera platform.

Use this endpoint after upgrading Feldera to rebuild pipelines that were compiled with older platform versions. In most cases, recompilation is not required—pipelines compiled with older versions will continue to run on the upgraded platform.

Situations where recompilation may be necessary: - To benefit from the latest bug fixes and performance optimizations. - When backward-incompatible changes are introduced in Feldera. In this case, attempting to start a pipeline compiled with an unsupported version will result in an error.

If the pipeline is already compiled with the current platform version, this operation is a no-op.

Note that recompiling the pipeline with a new platform version may change its query plan. If the modified pipeline is started from an existing checkpoint, it may require bootstrapping parts of its state from scratch. See Feldera documentation for details on the bootstrapping process.

Raises:

FelderaAPIError – If the pipeline is not in a STOPPED state.

version() → int[source]

Return the version of the pipeline.

views() → List[SQLView][source]

Return the views of the pipeline.

wait_for_completion(force_stop: bool = False, timeout_s: float | None = None, unavailable_timeout_s: float = 60.0)[source]

Block until the pipeline has completed processing all input records.

This method blocks until (1) all input connectors attached to the pipeline have finished reading their input data sources and issued end-of-input notifications to the pipeline, and (2) all inputs received from these connectors have been fully processed and corresponding outputs have been sent out through the output connectors.

This method will block indefinitely if at least one of the input connectors attached to the pipeline is a streaming connector, such as Kafka, that does not issue the end-of-input notification.

Parameters:

• force_stop – If True, the pipeline will be forcibly stopped after completion. False by default. No checkpoints will be made.

• timeout_s – Optional. The maximum time (in seconds) to wait for the pipeline to complete. The default is None, which means wait indefinitely.

• unavailable_timeout_s – The maximum time (in seconds) to wait for a pipeline in the UNAVAILABLE status to become available again. The manager reports UNAVAILABLE when one status probe fails, for example when the node is overloaded, so this status is usually transient. 60 seconds by default.

Raises:

RuntimeError – If the pipeline is not running, stays UNAVAILABLE for longer than unavailable_timeout_s, restarts during the wait without exactly-once fault tolerance, or returns unknown metrics.

wait_for_idle(idle_interval_s: float = 5.0, timeout_s: float | None = None, poll_interval_s: float = 0.2)[source]

Wait for the pipeline to become idle and then returns.

Idle is defined as a sufficiently long interval in which the number of input and processed records reported by the pipeline do not change, and they equal each other (thus, all input records present at the pipeline have been processed).

Parameters:

• idle_interval_s – Idle interval duration (default is 5.0 seconds).

• timeout_s – Timeout waiting for idle (None = no timeout is enforced).

• poll_interval_s – Polling interval, should be set substantially smaller than the idle interval (default is 0.2 seconds).

Raises:

• ValueError – If idle interval is larger than timeout, poll interval is larger than timeout, or poll interval is larger than idle interval.

• RuntimeError – If the metrics are missing or the timeout was reached.

wait_for_status(expected_status: PipelineStatus, timeout: int | None = None) → None[source]

Wait for the pipeline to reach the specified status.

Parameters:

• expected_status – The status to wait for

• timeout – Maximum time to wait in seconds. If None, waits forever (default: None)

Raises:

TimeoutError – If the expected status is not reached within the timeout

wait_for_token(token: str)[source]

Blocks until the pipeline processes all inputs represented by the completion token.

The pipeline has then handed the resulting changes to every output connector. A handle from listen() may still be receiving them; OutputHandler.wait_for_token() waits until it has.

feldera.enums module

class feldera.enums.BootstrapPolicy(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

ALLOW = 'allow'

AWAIT_APPROVAL = 'await_approval'

REJECT = 'reject'

static from_str(value)[source]

class feldera.enums.BuildMode(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

CREATE = 1

GET = 2

GET_OR_CREATE = 3

class feldera.enums.CheckpointStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Failure = 2

InProgress = 3

Success = 1

Unknown = 4

__init__(value)[source]

get_error() → str | None[source]

Returns the error, if any.

class feldera.enums.CompilationProfile(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

The compilation profile to use when compiling the program.

DEV = 'dev'

The development compilation profile.

OPTIMIZED = 'optimized'

The optimized compilation profile, the default for this API.

OPTIMIZED_SYMBOLS = 'optimized_symbols'

The optimized symbols compilation profile, good for profiling and debugging.

SERVER_DEFAULT = None

The compiler server default compilation profile.

UNOPTIMIZED = 'unoptimized'

The unoptimized compilation profile.

class feldera.enums.CompletionTokenStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

COMPLETE = 'complete'

Feldera has completed processing all inputs represented by this token.

IN_PROGRESS = 'inprogress'

Feldera is still processing the inputs represented by this token.

class feldera.enums.DeploymentDesiredStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Deployment desired status of the pipeline.

PAUSED = 3

RUNNING = 4

STANDBY = 2

STOPPED = 0

SUSPENDED = 5

UNAVAILABLE = 1

static from_str(value)[source]

class feldera.enums.DeploymentResourcesDesiredStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

The desired status of deployment resources of the pipeline.

PROVISIONED = 1

STOPPED = 0

static from_str(value)[source]

class feldera.enums.DeploymentResourcesStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

The desired status of deployment resources of the pipeline.

PROVISIONED = 2

PROVISIONING = 1

STOPPED = 0

STOPPING = 3

static from_str(value)[source]

class feldera.enums.DeploymentRuntimeDesiredStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Deployment runtime desired status of the pipeline.

PAUSED = 2

RUNNING = 3

STANDBY = 1

SUSPENDED = 4

UNAVAILABLE = 0

static from_str(value)[source]

class feldera.enums.DeploymentRuntimeStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Deployment runtime status of the pipeline.

AWAITINGAPPROVAL = 2

BOOTSTRAPPING = 4

CONCURRENTBOOTSTRAPPING = 9

INITIALIZING = 3

PAUSED = 6

REPLAYING = 5

RUNNING = 7

STANDBY = 1

SUSPENDED = 8

SYNCHRONIZING = 10

UNAVAILABLE = 0

static from_str(value)[source]

class feldera.enums.FaultToleranceModel(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

The fault tolerance model.

AtLeastOnce = 1

Each record is output at least once. Crashes may duplicate output, but no input or output is dropped.

ExactlyOnce = 2

Each record is output exactly once. Crashes do not drop or duplicate input or output.

static from_str(value)[source]

class feldera.enums.PipelineFieldSelector(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

ALL = 'all'

Select all fields of a pipeline.

STATUS = 'status'

Select only the fields required to know the status of a pipeline.

class feldera.enums.PipelineStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Represents the state that this pipeline is currently in.

AWAITINGAPPROVAL = 5

BOOTSTRAPPING = 7

CONCURRENTBOOTSTRAPPING = 13

INITIALIZING = 6

NOT_FOUND = 0

PAUSED = 9

PROVISIONING = 2

REPLAYING = 8

RUNNING = 10

STANDBY = 4

STOPPED = 1

STOPPING = 12

SUSPENDED = 11

SYNCHRONIZING = 14

UNAVAILABLE = 3

static from_str(value)[source]

class feldera.enums.ProgramStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

CompilingRust = 4

CompilingSql = 2

Pending = 1

RustError = 7

SqlCompiled = 3

SqlError = 6

Success = 5

SystemError = 8

__init__(value)[source]

static from_value(value)[source]

get_error() → dict | None[source]

Returns the compilation error, if any.

class feldera.enums.StorageStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Represents the current storage usage status of the pipeline.

CLEARED = 0

The pipeline has not been started before, or the user has cleared storage.

In this state, the pipeline has no storage resources bound to it.

CLEARING = 2

The pipeline is in the process of becoming unbound from its storage resources.

If storage resources are configured to be deleted upon clearing, their deletion occurs before transitioning to CLEARED. Otherwise, no actual work is required, and the transition happens immediately.

If storage is not deleted during clearing, the responsibility to manage or delete those resources lies with the user.

INUSE = 1

The pipeline was (attempted to be) started before, transitioning from STOPPED to PROVISIONING, which caused the storage status to become INUSE.

Being in the INUSE state restricts certain edits while the pipeline is STOPPED.

The pipeline remains in this state until the user invokes /clear, transitioning it to CLEARING.

static from_str(value)[source]

class feldera.enums.TransactionStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Represents the transaction handling status of a pipeline.

CommitInProgress = 3

A commit is currently in progress.

NoTransaction = 1

There is currently no active transaction.

TransactionInProgress = 2

There is an active transaction in progress.

static from_str(value)[source]

feldera.output_handler module

class feldera.output_handler.OutputHandler(client: FelderaClient, pipeline_name: str, view_name: str)[source]

Bases: object

__init__(client: FelderaClient, pipeline_name: str, view_name: str)[source]

Initializes the output handler, but doesn’t start it. To start the output handler, call the .OutputHandler.start method.

start()[source]

Starts the output handler in a separate thread

to_dict(clear_buffer: bool = True)[source]

Returns the output of the pipeline as a list of python dictionaries

Parameters:

clear_buffer – Whether to clear the buffer after getting the output.

to_pandas(clear_buffer: bool = True)[source]

Returns the output of the pipeline as a pandas DataFrame

Parameters:

clear_buffer – Whether to clear the buffer after getting the output.

wait_for_token(token: str, timeout_s: float | None = None)[source]

Blocks until this handle has received every change that the completion token covers.

Pipeline.wait_for_token() returns once the pipeline has handed the changes to the view’s output connectors; the changes can still be on their way to this handle. This method waits until the handle has received them, so to_pandas() afterwards returns every change for the token, along with any later change that may have arrived. If the input caused no view changes, the method still waits until the stream has passed the token’s step; an empty result received afterwards means that the input caused no view changes.

The handle receives only changes made after it started listening, so create it before sending the input. For a token that completed before that, this method returns only at the pipeline’s next step.

Parameters:

• token – A completion token, as returned by Pipeline.input_json() or Pipeline.input_pandas().

• timeout_s – The amount of time in seconds to wait; None waits without limit.

Raises:

FelderaTimeoutError – when the changes have not arrived after timeout_s seconds.

feldera.runtime_config module

class feldera.runtime_config.Resources(config: Mapping[str, Any] | None = None, cpu_cores_max: int | None = None, cpu_cores_min: int | None = None, memory_mb_max: int | None = None, memory_mb_min: int | None = None, storage_class: str | None = None, volume_attributes_class: str | None = None, storage_mb_max: int | None = None, memory_max: str | None = None, memory_min: str | None = None, storage_max: str | None = None, storage_min: str | None = None, storage_mb_min: int | None = None)[source]

Bases: object

Class used to specify the resource configuration for a pipeline.

Memory and storage sizes are strings made of a number and a unit, such as "4Gi", "16G" or "400Gi". The deprecated integer arguments count megabytes and emit a DeprecationWarning.

Parameters:

• config – A dictionary containing all the configuration values.

• cpu_cores_max – The maximum number of CPU cores to reserve for an instance of the pipeline.

• cpu_cores_min – The minimum number of CPU cores to reserve for an instance of the pipeline.

• memory_max – The maximum memory to reserve for an instance of the pipeline, such as "16Gi".

• memory_min – The minimum memory to reserve for an instance of the pipeline, such as "4Gi".

• storage_class – The storage class to use for the pipeline. The class determines storage performance such as IOPS and throughput.

• volume_attributes_class – The Kubernetes volume attributes class to use for the pipeline storage. The class sets mutable storage parameters such as IOPS and throughput.

• storage_max – The storage to reserve for an instance of the pipeline, such as "400Gi".

• storage_min – The initial storage of an instance of the pipeline; set with storage_max to enable storage autoscaling.

• memory_mb_max – Deprecated; use memory_max.

• memory_mb_min – Deprecated; use memory_min.

• storage_mb_max – Deprecated; use storage_max.

• storage_mb_min – Deprecated; use storage_min.

__init__(config: Mapping[str, Any] | None = None, cpu_cores_max: int | None = None, cpu_cores_min: int | None = None, memory_mb_max: int | None = None, memory_mb_min: int | None = None, storage_class: str | None = None, volume_attributes_class: str | None = None, storage_mb_max: int | None = None, memory_max: str | None = None, memory_min: str | None = None, storage_max: str | None = None, storage_min: str | None = None, storage_mb_min: int | None = None)[source]

property memory_max: str | None

memory_max, a size such as "4Gi".

property memory_mb_max: int | None

Deprecated: memory_max as a whole number of the old unit.

property memory_mb_min: int | None

Deprecated: memory_min as a whole number of the old unit.

property memory_min: str | None

memory_min, a size such as "4Gi".

property storage_max: str | None

storage_max, a size such as "4Gi".

property storage_mb_max: int | None

Deprecated: storage_max as a whole number of the old unit.

property storage_mb_min: int | None

Deprecated: storage_min as a whole number of the old unit.

property storage_min: str | None

storage_min, a size such as "4Gi".