Submodules

feldera.rest.config module

class feldera.rest.config.Config(url: str | None = None, api_key: str | Callable[[], str] | None = None, version: str | None = None, timeout: float | None = None, connection_timeout: float | None = None, requests_verify: bool | str | None = None, retry_config: RetryConfig | None = None, tenant: str | None = None, headers: Mapping[str, str] | None = None)[source]

Bases: object

FelderaClient configuration, which includes authentication information and the address of the Feldera API the client will interact with.

__init__(url: str | None = None, api_key: str | Callable[[], str] | None = None, version: str | None = None, timeout: float | None = None, connection_timeout: float | None = None, requests_verify: bool | str | None = None, retry_config: RetryConfig | None = None, tenant: str | None = None, headers: Mapping[str, str] | None = None) → None[source]

See documentation of the FelderaClient constructor for the other arguments.

Parameters:

• version – (Optional) Version of the API to use. Default: v0.

• retry_config – (Optional) Retry behavior for transient HTTP failures. Default: RetryConfig() — 3 retries with exponential backoff starting at 2 seconds.

• tenant – (Optional) Tenant to act in, sent as the Feldera-Tenant header. A platform owner uses this to select any tenant (by name or UUID); a regular user, to disambiguate among the tenants their token authorizes. Default: the token’s own/home tenant.

• headers – (Optional) Extra HTTP headers sent with every request. A header given here replaces the one the client would otherwise send under that name, whatever case either spells it in, except Content-Type, which each request sets itself. Default: no extra headers.

feldera.rest.errors module

exception feldera.rest.errors.FelderaAPIError(error: str, request: Response)[source]

Bases: FelderaError

Error sent by Feldera API

__init__(error: str, request: Response) → None[source]

feldera.rest.errors.FelderaApiError

alias of FelderaAPIError

exception feldera.rest.errors.FelderaCommunicationError(err: str)[source]

Bases: FelderaError

Error when connection to Feldera

__init__(err: str) → None[source]

exception feldera.rest.errors.FelderaError(message: str)[source]

Bases: Exception

Generic class for Feldera error handling

__init__(message: str) → None[source]

exception feldera.rest.errors.FelderaTimeoutError(err: str)[source]

Bases: FelderaError

Error when Feldera operation takes longer than expected

__init__(err: str) → None[source]

exception feldera.rest.errors.TableReloadNotAllowedError(pipeline_name: str, reason: str, approval_diff: Mapping[str, Any] | None, blocked_tables: List[str] | None = None)[source]

Bases: RuntimeError

Raised when bootstrapping would reload a table that is not in reloadable_tables, or when the SDK cannot tell which tables it reloads. The pipeline stays in the AWAITINGAPPROVAL state: call approve() to bootstrap or stop(force=True) to reject.

__init__(pipeline_name: str, reason: str, approval_diff: Mapping[str, Any] | None, blocked_tables: List[str] | None = None) → None[source]

Parameters:

• pipeline_name – The name of the pipeline. Stored as pipeline_name.

• reason – Why the SDK did not approve the bootstrap. Part of the error message.

• approval_diff – The diff awaiting approval, as the pipeline reported it. Stored as approval_diff.

• blocked_tables – Modified tables that are not in reloadable_tables. Stored as blocked_tables. Empty if the diff is unknown.

feldera.rest.feldera_client module

class feldera.rest.feldera_client.FelderaClient(url: str | None = None, api_key: str | Callable[[], str] | None = None, timeout: float | None = None, connection_timeout: float | None = None, requests_verify: bool | str | None = None, retry_config: RetryConfig | None = None, tenant: str | None = None, headers: Mapping[str, str] | None = None)[source]

Bases: object

A client for the Feldera HTTP API.

The client is initialized with the configuration needed for interacting with the Feldera HTTP API, which it uses in its calls. Its methods are implemented by issuing one or more HTTP requests to the API, and as such can provide higher level operations (e.g., support waiting for the success of asynchronous HTTP API functionality).

__init__(url: str | None = None, api_key: str | Callable[[], str] | None = None, timeout: float | None = None, connection_timeout: float | None = None, requests_verify: bool | str | None = None, retry_config: RetryConfig | None = None, tenant: str | None = None, headers: Mapping[str, str] | None = None) → None[source]

Constructs a Feldera client.

Parameters:

• url – (Optional) URL to the Feldera API. The default is read from the FELDERA_HOST environment variable; if the variable is not set, the default is “http://localhost:8080”.

• api_key –

  (Optional) Bearer credential. Either: - a string — a Feldera API key (“apikey:…”) or a long-lived JWT; - a zero-arg callable returning a string — resolved before every

  request, suitable for short-lived OIDC tokens (Kubernetes projected SA token, GitHub Actions OIDC, Tailscale tsidp, …). On HTTP 401 the callable is re-invoked once and the request retried, so a token expiring mid-flight in a long-running script self-heals.

  The default is read from the FELDERA_API_KEY environment variable; if the variable is not set, the default is None (no API key is provided).

• timeout – (Optional) Duration in seconds that the client will wait to receive a response of an HTTP request, after which it times out. The default is None (wait indefinitely; no timeout is enforced).

• connection_timeout – (Optional) Duration in seconds that the client will wait to establish the connection of an HTTP request, after which it times out. The default is None (wait indefinitely; no timeout is enforced).

• requests_verify – (Optional) The verify parameter passed to the requests library, which is used internally to perform HTTP requests. See also: https://requests.readthedocs.io/en/latest/user/advanced/#ssl-cert-verification . The default is based on the FELDERA_HTTPS_TLS_CERT or the FELDERA_TLS_INSECURE environment variable. By setting FELDERA_HTTPS_TLS_CERT to a path, the default becomes the CA bundle it points to. By setting FELDERA_TLS_INSECURE to “1”, “true” or “yes” (all case-insensitive), the default becomes False which means to disable TLS verification by default. If both variables are set, the former takes priority over the latter. If neither variable is set, the default is True.

• retry_config – (Optional) Retry behavior for transient HTTP failures (408, 502, 503, 504, timeouts). The default is RetryConfig() — 3 retries with exponential backoff starting at 2 seconds.

• tenant – (Optional) Tenant to act in, sent as the Feldera-Tenant header. A platform owner uses this to select any tenant (by name or UUID); a regular user, to disambiguate among the tenants their token authorizes. The default is read from FELDERA_TENANT; if unset, the server uses the token’s own tenant.

• headers –

  (Optional) Extra HTTP headers sent with every request, as a mapping of name to value. A header given here replaces the one the client would otherwise send under that name, whatever case either spells it in. Content-Type is the exception: each request sets its own, because it describes the body the client serialized.

  Use it to reach a deployment behind an authenticating proxy, which admits a request by its own session cookie rather than by a Feldera API key:

  FelderaClient(url, headers={"Cookie": alb_session_cookie})
  

  The default is no extra headers.

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

Parameters:

• pipeline_name – The name of the pipeline to activate

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

• timeout_s – The amount of time in seconds to wait for the pipeline to activate.

add_tenant_member(subject: str, role: str, email: str | None = None) → dict[source]

Add a member to the current tenant by identity, before the user’s first login. Requires the admin role.

Parameters:

• subject – The user’s OIDC subject (the sub claim of their tokens).

• role – The role to grant: "read", "write" or "admin". It may not exceed the caller’s own role.

• email – (Optional) Email shown in the member list.

Returns:

A dict with the new member’s user_id.

advance_clock(pipeline_name: str, 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 the pipeline. The clock is forward-only.

Parameters:

• pipeline_name – The name of the pipeline.

• 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). Must be non-negative. Non-zero values round 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.

approve_pipeline(pipeline_name: str, silent_bootstrap: bool = False, concurrent_bootstrap: bool = False)[source]

Approve a pipeline awaiting approval to proceed with bootstrapping.

Parameters:

• pipeline_name – The name of the pipeline to approve.

• 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.

checkpoint_pipeline(pipeline_name: str) → int[source]

Triggers checkpointing a pipeline and returns just the checkpoint sequence number.

This is for backward compatibility: checkpoint_pipeline_response is a better choice because it allows the caller to detect that the pipeline has restarted (and that therefore the checkpoint needs to be triggered again).

checkpoint_pipeline_response(pipeline_name: str) → dict[source]

Triggers checkpointing a pipeline, returning the full response.

Parameters:

pipeline_name – The name of the pipeline to checkpoint

Returns:

dict with checkpoint_sequence_number and incarnation_uuid, the latter identifying the pipeline process instance that accepted the request; pass it to checkpoint_pipeline_status to detect a pipeline restart between the request and the status check.

checkpoint_pipeline_status(pipeline_name: str, incarnation_uuid: str | None = None) → dict[source]

Gets the checkpoint status

Parameters:

• pipeline_name – The name of the pipeline to check the checkpoint status of.

• incarnation_uuid – The incarnation_uuid returned by the checkpoint_pipeline_response call, to allow pipeline restarts to be clearly reported as IncarnationUuidMismatch.

clear_storage(pipeline_name: str, 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:

• pipeline_name – Name of the pipeline.

• 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(pipeline_name: str, 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]

Commits the currently active transaction.

Parameters:

• pipeline_name – The name of the pipeline.

• 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_status(pipeline_name: str, token: str) → Mapping[str, Any][source]

Returns the status of a completion token as the pipeline reports it: a mapping with status, either complete or inprogress, and step, the number of steps the pipeline completes before the token is complete, or None while the token’s records wait to be ingested.

Parameters:

• pipeline_name – The name of the pipeline

• token – The token to check for completion

completion_token_processed(pipeline_name: str, token: str) → bool[source]

Check whether the pipeline has finished processing all inputs received from the connector before the token was generated.

Parameters:

• pipeline_name – The name of the pipeline

• token – The token to check for completion

Returns:

True if the pipeline has finished processing all inputs represented by the token, False otherwise

create_api_key(name: str, role: str = 'write') → dict[source]

Create a new API key with the specified name and role.

The generated API key is returned in the response and cannot be retrieved again later, so store it securely.

Parameters:

• name – The API key name to create.

• role – The role the key carries, "read" or "write". The role may not exceed the caller’s own role, and admin/owner are never issuable as API keys. Defaults to "write" to preserve the read+write access earlier SDK versions granted implicitly.

Returns:

A dict with keys: id (UUID string), name, and api_key.

Raises:

FelderaAPIError – If a key with the same name already exists, or the requested role exceeds the caller’s role.

create_oidc_trust(name: str, issuer: str, subject: str, audience: str | None = None, description: str | None = None, role: str | None = None) → dict[source]

Register a new OIDC trust relationship.

Any JWT signed by issuer whose sub claim matches subject (and, if specified, aud claim matches audience) is authorized to act in the current tenant with the granted role. * is a wildcard in subject and audience.

Parameters:

• name – Unique name within the tenant.

• issuer – Issuer URL exactly as it appears in the iss claim.

• subject – Pattern matched against the JWT sub claim.

• audience – Pattern matched against the JWT aud claim. Omit to skip audience matching.

• description – Free-text description.

• role – Role granted to a matching token: read, write or admin, capped at the caller’s role. Defaults to read server-side when omitted.

Returns:

A dict with keys id and name.

Raises:

FelderaAPIError – If name is already in use or fields are invalid.

create_or_update_pipeline(pipeline: Pipeline, wait: bool = True) → Pipeline[source]

Create a pipeline if it doesn’t exist or update a pipeline and wait for it to compile

Parameters:

• pipeline – The pipeline to create or update

• wait – Whether to wait for the pipeline to compile. True by default

Returns:

The created or updated pipeline

create_pipeline(pipeline: Pipeline, wait: bool = True) → Pipeline[source]

Create a pipeline if it doesn’t exist and wait for it to compile

Parameters:

• pipeline – The pipeline to create

• wait – Whether to wait for the pipeline to compile. True by default

create_tenant(name: str) → dict[source]

Create a tenant, or return it if one with this name already exists. Platform owners only.

A login resolves its tenant by name, so a user whose identity provider asserts this name lands in the tenant created here.

Parameters:

name – The name of the tenant.

Returns:

A dict describing the tenant (id, name, initial_provider).

delete_api_key(name: str) → None[source]

Delete an API key by name, which revokes it.

Parameters:

name – The API key name.

Raises:

FelderaAPIError – If there is no API key with this name.

delete_oidc_trust(name: str) → None[source]

Delete an OIDC trust relationship by name.

delete_pipeline(name: str)[source]

Deletes a pipeline by name

Parameters:

name – The name of the pipeline

delete_tenant(tenant_id: str) → None[source]

Delete a tenant. Platform owners only.

The tenant must hold no pipelines, API keys or OIDC trust relationships; delete those first.

Parameters:

tenant_id – Identifier of the tenant to delete.

Raises:

FelderaAPIError – If the tenant still holds any of the above.

dismiss_error_pipeline(pipeline_name: str)[source]

Dismisses the deployment error of a pipeline.

Parameters:

pipeline_name – The name of the pipeline for which to dismiss the deployment error

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

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

Parameters:

• pipeline_name – The name of the pipeline

• table_name – The name of the table associated with this connector.

• connector_name – The name of the connector.

Raises:

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

get_api_key(name: str) → dict[source]

Retrieve the metadata of an API key by name.

Parameters:

name – The API key name.

Returns:

A dict with keys id, name and role.

Raises:

FelderaAPIError – If there is no API key with this name.

get_checkpoints(pipeline_name: str)[source]

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

Retrieves a specific cluster 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).

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

Retrieves all cluster events (status fields only).

Returns:

List of cluster events.

get_config() → FelderaConfig[source]

Retrieves the general Feldera server configuration.

get_oidc_trust(name: str) → dict[source]

Retrieve a single OIDC trust relationship by name.

Parameters:

name – Trust relationship name.

get_pipeline(pipeline_name: str, field_selector: PipelineFieldSelector) → Pipeline[source]

Get a pipeline by name

Parameters:

• pipeline_name – The name of the pipeline

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

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

Retrieves a specific pipeline event.

Parameters:

• pipeline_name – Name of the pipeline.

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

get_pipeline_events(pipeline_name: str, *, selector: str = 'status') → list[dict][source]

Retrieves all pipeline events (status fields only).

Parameters:

• pipeline_name – Name of the pipeline.

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

Returns:

List of pipeline events.

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

Get the pipeline logs

Parameters:

• name – The name of the pipeline

• follow – Set False to stop after the retained logs instead of waiting for new lines. A server older than this parameter ignores it.

Returns:

A generator yielding the logs, one line at a time.

get_pipeline_metrics(name: str, format: str = 'prometheus')[source]

Get the pipeline circuit metrics

Parameters:

• name – The name of the pipeline

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

Returns:

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

get_pipeline_stats(name: str) → dict[source]

Get the pipeline metrics and performance counters

Parameters:

name – The name of the pipeline

get_pipeline_support_bundle(pipeline_name: str, params: Dict[str, Any] | None = None, retry: bool = False) → bytes[source]

Generate a support bundle containing diagnostic information from a pipeline.

This endpoint 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:

• pipeline_name – The name of the pipeline

• params – Optional query parameters to control data collection

• retry – Whether to retry after a failure that can come late, such as a gateway timeout or a dropped connection. A bundle that collects new data is slow and loads the pipeline, so pass True only with collect=false in params.

Returns:

The support bundle as bytes (ZIP file)

Raises:

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

get_remote_checkpoints(pipeline_name: str) → list[dict][source]

List checkpoints available in the configured remote object storage.

get_runtime_config(pipeline_name) → Mapping[str, Any][source]

Get the runtime config of a pipeline by name

Parameters:

pipeline_name – The name of the pipeline

get_samply_profile(pipeline_name: str) → bytes[source]

Get the most recent available samply profile for the given pipeline.

Parameters:

pipeline_name – The name of the pipeline

Returns:

The samply profile as bytes (GZIP file)

Raises:

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

get_tenant(tenant: str) → dict[source]

Retrieve a single tenant by name or identifier. Platform owners only.

A selector that parses as a UUID is looked up by tenant identifier, otherwise by name.

Parameters:

tenant – The tenant’s name or identifier (UUID).

Returns:

A dict describing the tenant (id, name, initial_provider).

Raises:

FelderaAPIError – If no tenant matches the selector.

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

Get the status of the specified input connector.

list_api_keys() → List[dict][source]

List the API keys of the current tenant.

The key itself is only ever returned by create_api_key().

Returns:

A list of dicts with keys id, name and role.

list_oidc_trust() → List[dict][source]

List the OIDC trust relationships configured for the current tenant.

Returns:

A list of dicts each describing a trust relationship (id, name, description, issuer, subject, audience, role).

list_tenant_members() → List[dict][source]

List the members of the current tenant and their roles. Requires the admin role.

Returns:

A list of dicts each describing a member (user_id, provider, subject, role, and, when known, email, email_verified, display_name and origin).

list_tenants() → List[dict][source]

List every tenant in the installation. Platform owners only.

Returns:

A list of dicts each describing a tenant (id, name, initial_provider).

listen_to_pipeline(pipeline_name: str, table_name: str, format: str, send_snapshot: bool | None = None, backpressure: bool = True, array: bool = False, timeout: float | None = None, case_sensitive: bool = False)[source]

Listen for updates to views for pipeline, yields the chunks of data

Parameters:

• pipeline_name – The name of the pipeline

• table_name – The name of the table to listen to

• format – The format of the data, either “json” or “csv”

• send_snapshot – When True, the connector delivers a full snapshot of the materialized view before streaming incremental updates. The view must be materialized. Defaults to False.

• backpressure – When the flag is True (the default), this method waits for the consumer to receive each chunk and blocks the pipeline if the consumer cannot keep up. When this flag is False, the pipeline drops data chunks if the consumer is not keeping up with its output. This prevents a slow consumer from slowing down the entire pipeline.

• array – Set True to group updates in this stream into JSON arrays, used in conjunction with the “json” format, the default value is False

• timeout – The amount of time in seconds to listen to the stream for

• case_sensitive – True if the table name is case sensitive or a reserved keyword, False by default

static localhost(port: int = 8080) → FelderaClient[source]

Create a FelderaClient that connects to the local Feldera instance

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

Get the status of the specified output connector.

patch_pipeline(name: str, 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]

Incrementally update pipeline

Parameters:

name – The name of the pipeline

Each field is patched independently: a provided value overwrites the stored one, and a field left as None is omitted from the request and leaves the stored value untouched. To clear the description or tags, pass an empty string or empty list, respectively.

Tags, when provided, are normalized before being sent: color variants of the same display name are collapsed and the result is stored in API server database in lexicographic order (see feldera.tags).

pause_connector(pipeline_name, table_name, connector_name)[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(pipeline_name: str, 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:

• pipeline_name – The name of the pipeline.

• table_name – The name of the table associated with this connector.

• connector_name – The name of the connector.

Raises:

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

pause_output_connector(pipeline_name: str, 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:

• pipeline_name – The name of the pipeline.

• view_name – The name of the view associated with this connector.

• connector_name – The name of the connector.

Raises:

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

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

Pause a pipeline

Parameters:

• pipeline_name – The name of the pipeline to stop

• error_message – The error message to show if the pipeline is in STOPPED state due to a failure.

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

• timeout_s – The amount of time in seconds to wait for the pipeline to pause.

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

Compute the diff between the 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 the 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:

• pipeline_name – The name of the pipeline.

• 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.

pipelines(selector: PipelineFieldSelector = PipelineFieldSelector.STATUS) → list[Pipeline][source]

Get all pipelines

push_to_pipeline(pipeline_name: str, table_name: str, format: str, data: list[list | str | dict] | dict | str, array: bool = False, force: bool = False, update_format: str = 'raw', json_flavor: str | None = None, serialize: bool = True, wait: bool = True, wait_timeout_s: float | None = None, connector_metadata: Mapping[str, Any] | None = None) → str[source]

Insert data into a pipeline

Parameters:

• pipeline_name – The name of the pipeline

• table_name – The name of the table

• format – The format of the data, either “json” or “csv”

• array – True if updates in this stream are packed into JSON arrays, used in conjunction with the “json” format

• force – If True, the data will be inserted even if the pipeline is paused

• update_format – JSON data change event format, used in conjunction with the “json” format, the default value is “insert_delete”, other supported formats: “debezium”, “raw”

• json_flavor – JSON encoding used for individual table records, the default value is “default”, other supported encodings: “debezium_mysql”, “snowflake”, “kafka_connect_json_converter”, “pandas”

• data – The data to insert

• serialize – If True, the data will be serialized to JSON. True by default

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

• wait_timeout_s – The timeout in seconds to wait for this set of inputs to be processed by the pipeline. None by default

• 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. None by default.

Returns:

The completion token to this input.

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

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

Arrow IPC preserves SQL type fidelity that the JSON format cannot: integers wider than 53 bits, MAP keys that are not strings, and result sets where two columns share a name. Prefer this method over query_as_json() when programmatically inspecting results.

Parameters:

• pipeline_name – The name of the pipeline to query.

• query – The SQL query to be executed.

Returns:

A generator that yields each query batch as a pyarrow.RecordBatch.

query_as_hash(pipeline_name: str, query: str) → str[source]

Executes an ad-hoc query on the specified pipeline and returns a hash of the result.

Parameters:

• pipeline_name – The name of the pipeline to query.

• query – The SQL query to be executed.

Returns:

A string containing the hash of the query result.

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

Executes an ad-hoc query on the specified pipeline and returns the result as a generator that yields rows of the query as Python dictionaries.

Finite floating-point numbers are decoded with decimal.Decimal so the digits printed on the JSON line round-trip exactly. JSON cannot represent NaN, Infinity or -Infinity; the server emits those as null and they arrive here as None.

Deprecated since version The: JSON format coerces numbers to f64 (losing precision for wide integers), drops NaN and infinities, cannot represent SQL MAP keys that are not strings, and silently drops result columns that share a name. Use query_as_arrow() instead. See https://github.com/feldera/feldera/issues/4219.

Parameters:

• pipeline_name – The name of the pipeline to query.

• query – The SQL query to be executed.

Returns:

A generator that yields each row of the result as a Python dictionary, deserialized from JSON.

query_as_parquet(pipeline_name: str, query: str, path: str)[source]

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

Parameters:

• pipeline_name – The name of the pipeline to query.

• query – The SQL query to be executed.

• path – The path including the file name to save the resulting parquet file in.

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

Executes an ad-hoc query on the specified pipeline and returns a generator that yields lines of the table.

Parameters:

• pipeline_name – The name of the pipeline to query.

• query – The SQL query to be executed.

Returns:

A generator yielding the query result in tabular format, one line at a time.

rebalance_pipeline(pipeline_name: str)[source]

remove_tenant_member(user_id: str) → None[source]

Remove a member from the current tenant. Requires the admin role.

This drops their role now, but if the identity provider still grants them the tenant, they are added again on their next login. Revoke access at the provider to lock them out completely.

Parameters:

user_id – The member’s user_id, as listed by list_tenant_members().

Raises:

FelderaAPIError – If the user is not a member of the tenant.

rename_tenant(tenant_id: str, name: str, displace_existing: bool = False) → dict[source]

Rename a tenant. Platform owners only.

A login resolves its tenant by name, so the new name decides which users arrive in this tenant. Renaming a tenant to a name its provider no longer asserts sends those users to a new, empty tenant on their next request.

Parameters:

• tenant_id – Identifier of the tenant to rename.

• name – The new name.

• displace_existing – Take the name from the tenant that currently holds it, which is renamed to <name> (<id>) and keeps everything it had. Needed to recover a tenant no login reaches, because every request re-creates the name its token resolves.

Returns:

A dict whose displaced key describes the tenant that gave up the name, or is None when the name was free.

Raises:

FelderaAPIError – If the name is taken and displace_existing is not set.

resume_connector(pipeline_name: str, 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_pipeline(pipeline_name: str, wait: bool = True, timeout_s: float | None = None)[source]

Resume a pipeline

Parameters:

• pipeline_name – The name of the pipeline to stop

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

• timeout_s – The amount of time in seconds to wait for the pipeline to pause.

resume_pipeline_logs(pipeline_name: str, cursor: str | None = None) → LogStream[source]

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

Where get_pipeline_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. A reader is therefore never locked out of its logs by a stale cursor.

Parameters:

• pipeline_name – The name of the pipeline.

• 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.

set_tenant_member_role(user_id: str, role: str) → None[source]

Change a member’s role in the current tenant. Requires the admin role.

Parameters:

• user_id – The member’s user_id, as listed by list_tenant_members().

• role – The new role: "read", "write" or "admin". It may not exceed the caller’s own role.

Raises:

FelderaAPIError – If the user is not a member of the tenant.

start_compaction_pipeline(pipeline_name: str)[source]

start_input_connector(pipeline_name: str, 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:

• pipeline_name – The name of the pipeline.

• table_name – The name of the table associated with this connector.

• connector_name – The name of the connector.

Raises:

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

start_output_connector(pipeline_name: str, 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:

• pipeline_name – The name of the pipeline.

• view_name – The name of the view associated with this connector.

• connector_name – The name of the connector.

Raises:

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

start_pipeline(pipeline_name: str, 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) → PipelineStatus | None[source]

Parameters:

• pipeline_name – The name of the pipeline to start

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

• timeout_s – The amount of time in seconds to wait for the pipeline to start.

• 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, with bootstrap_policy=BootstrapPolicy.ALLOW. See Pipeline.start.

start_pipeline_as_paused(pipeline_name: str, 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) → PipelineStatus | None[source]

Parameters:

• pipeline_name – The name of the pipeline to start as paused.

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

• timeout_s – The amount of time in seconds to wait for the pipeline to start.

• 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, with bootstrap_policy=BootstrapPolicy.ALLOW. See Pipeline.start.

start_pipeline_as_standby(pipeline_name: str, 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]

Parameters:

• pipeline_name – The name of the pipeline to start as standby.

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

• timeout_s – The amount of time in seconds to wait for the pipeline to start.

• 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, with bootstrap_policy=BootstrapPolicy.ALLOW. See Pipeline.start.

start_samply_profile(pipeline_name: str, duration: int)[source]

Start the profiling this pipeline with samply.

Parameters:

• pipeline_name – The name of the pipeline

• 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_transaction(pipeline_name: str) → int[source]

Start a new transaction and return its ID.

start_transaction_response is a better choice because it also returns the pipeline’s incarnation_uuid, which lets commit_transaction detect a pipeline restart.

Parameters:

pipeline_name – The name of the pipeline.

start_transaction_response(pipeline_name: str) → dict[source]

Start a new transaction, returning the full response.

Parameters:

pipeline_name – The name of the pipeline.

Returns:

dict with transaction_id and incarnation_uuid, the latter identifying the pipeline process instance that started the transaction; 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.

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

Stop a pipeline

Parameters:

• pipeline_name – The name of the pipeline to stop

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

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

• timeout_s – The amount of time in seconds to wait for the pipeline to stop.

sync_checkpoint(pipeline_name: str) → str[source]

Triggers a checkpoint synchronization and returns just the checkpoint UUID.

This is for backward compatibility: sync_checkpoint_response is a better choice because it allows the caller to detect that the pipeline has restarted (and that therefore the checkpoint synchronization needs to be triggered again).

sync_checkpoint_response(pipeline_name: str) → dict[source]

Triggers a checkpoint synchronization, returning the full response.

Parameters:

pipeline_name – Name of the pipeline whose checkpoint should be synchronized.

Returns:

dict with checkpoint_uuid and incarnation_uuid, the latter identifying the pipeline process instance that accepted the request; pass it to sync_checkpoint_status to detect a pipeline restart between the request and the status check.

sync_checkpoint_status(pipeline_name: str, incarnation_uuid: str | None = None) → dict[source]

Gets the checkpoint sync status of the pipeline

Parameters:

• pipeline_name – The name of the pipeline to check the checkpoint synchronization status of.

• incarnation_uuid – The incarnation_uuid returned by the sync_checkpoint_response call this status check is for, to allow pipeline restarts to be clearly reported as IncarnationUuidMismatch.

testing_force_update_platform_version(name: str, platform_version: str)[source]

update_pipeline_runtime(name: str)[source]

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

Parameters:

name – The name of the pipeline

validate_program(program_code: str, runtime_version: str | None = None, ir: bool = False) → Dict[str, Any][source]

Validate a SQL program by compiling it, without creating a pipeline or building the pipeline binary.

Reports SQL errors and warnings and the derived schema and connectors. Set ir to also return the program IR (dataflow).

A SQL compilation error is reported in the returned result (as SqlError), not raised. FelderaAPIError is raised only for a system-level failure, such as an invalid runtime version or the compiler service being unavailable.

Parameters:

• program_code – The SQL program to validate.

• runtime_version – Runtime version to compile the 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.

• ir – Also return the program IR (dataflow) in the result.

Returns:

The validation result as a dict (see the ValidateProgramResponse schema): one of {"Success": {...}}, {"SqlError": {...}}, or {"SystemError": {...}}.

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

Blocks until all records represented by this completion token have been processed.

Parameters:

• pipeline_name – The name of the pipeline

• token – The token to check for completion

• timeout_s – The amount of time in seconds to wait for the pipeline to process these records.

feldera.rest.feldera_config module

class feldera.rest.feldera_config.FelderaConfig(cfg: dict)[source]

Bases: object

General configuration of the current Feldera instance.

__init__(cfg: dict)[source]

class feldera.rest.feldera_config.FelderaEdition(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. Represents the Feldera edition between Enterprise and Open source.

ENTERPRISE = 'Enterprise'

The Enterprise version of Feldera.

ENTERPRISE_DEV = 'EnterpriseDev'

The Enterprise edition for development use.

OPEN_SOURCE = 'Open source'

The open source version of Feldera.

UNKNOWN = 'Unknown'

An edition this client does not recognize, reported by a newer server.

static from_value(value)[source]

is_enterprise()[source]

feldera.rest.logs module

Reading a pipeline’s logs stream from a cursor.

class feldera.rest.logs.LogPosition(epoch: str, seq: int, gap: int)[source]

Bases: object

Where a logs stream starts, as reported by the response that opens it.

__init__(epoch: str, seq: int, gap: int) → None

cursor(lines_read: int = 0) → str[source]

The cursor that resumes this stream after lines_read of its lines.

The body holds log lines and nothing else, one per sequence number, so a reader derives its position by counting rather than by inspecting what it receives.

epoch: str

Identifies the lifetime of the pipeline’s logs buffer, which is what gives the sequence number meaning: the buffer lives in memory, so a runner restart resets numbering to zero while a reader still holds a cursor issued before it.

The buffer outlives a pipeline run: stopping and starting a pipeline keeps the epoch and continues the numbering, so only a runner restart or a deleted pipeline ends it.

static from_headers(headers: Mapping[str, str]) → LogPosition[source]

The position a logs response reports.

All three headers are required. A response carrying only some of them could not be turned into a cursor, so it is an error rather than something to interpret.

gap: int

Lines discarded between the requested cursor and seq, and which this stream will therefore never deliver. Zero means the resume is exact.

seq: int

Sequence number of the line preceding the stream’s first line.

class feldera.rest.logs.LogStream(response: Response, position: LogPosition)[source]

Bases: object

An open logs stream: where it starts, the lines that follow, and how far they reach.

Iterating yields one log line at a time, blocking for the next one until the pipeline is deleted or the connection drops. Close the stream when done reading, or use it as a context manager, which closes it on the way out.

Pass cursor() to the next connection to carry on from the line this one reached.

__init__(response: Response, position: LogPosition) → None[source]

close() → None[source]

cursor() → str[source]

The cursor that resumes the log after the last line this stream yielded.

The stream counts the lines it delivers, so a reader resumes by handing this back rather than by keeping a count of its own.

property lines_read: int

Log lines this stream has yielded.

feldera.rest.pipeline module

class feldera.rest.pipeline.Pipeline(name: str, sql: str, udf_rust: str, udf_toml: str, program_config: Mapping[str, Any], runtime_config: Mapping[str, Any], description: str | None = None, tags: List[str] | None = None)[source]

Bases: object

Represents a Feldera pipeline

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

Initializes a new pipeline

Parameters:

• name – The name of the pipeline

• sql – The SQL code of the pipeline

• udf_rust – Rust code for UDFs

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

• program_config – The program config of the pipeline

• runtime_config – The configuration of the pipeline

• description – Optional. The description of the pipeline

• tags – Optional. Free-form labels used to organize, group, and filter pipelines (e.g. “team:ml”, “env:prod”)

classmethod from_dict(d: Mapping[str, Any])[source]

feldera.rest.retry module

class feldera.rest.retry.RetryConfig(max_retries: int = 3, deadline_seconds: float | None = None, initial_backoff: float = 2.0, max_backoff: float = 64.0, multiplier: float = 2.0, jitter: float = 0.0, unhealthy_backoff: float = 90.0, expired_token_wait_seconds: float = 0.0, retryable_status_codes: ~typing.FrozenSet[int] = <factory>)[source]

Bases: object

Configures retry behavior for the Feldera HTTP client.

Retries are attempted on transient failures: connection/read timeouts and the HTTP statuses listed in retryable_status_codes (408, 429, 502, 503, 504 by default). 401 is not among them; it is handled separately, and only for a callable credential, by expired_token_wait_seconds.

A request that is not idempotent (POST/PUT/PATCH/DELETE not marked idempotent=True) retries only connect timeouts, 408/429, and a 503 whose error code shows the request was not applied (such as DeadlockDetected), because after other failures the server may already have applied it.

Wait strategies:

• 408, 429, 503, 504 and connection/read timeouts use exponential backoff: min(initial_backoff * (multiplier ** n), max_backoff), plus a uniform random [0, jitter) term, where n is the zero-based retry index.

• 502 uses cluster-aware backoff: the client probes /cluster_healthz; if the cluster is healthy, the 502 is treated as spurious and the next retry runs immediately (wait = 0). If the cluster reports unhealthy (e.g. an upgrade is in progress), the next retry waits unhealthy_backoff seconds.

• A server-supplied Retry-After header always overrides the computed wait (capped at max_backoff).

Parameters:

• max_retries – Number of retries to attempt after the initial request. A value of 3 means up to 4 total attempts. Must be >= 0. Ignored when deadline_seconds is set. Default: 3.

• deadline_seconds – Wall-clock retry budget in seconds, measured from the first attempt. When set, max_retries no longer applies: retries continue (with the usual waits) until the budget is exhausted. Transient outages such as a node replacement last for a duration, not a number of requests, so a duration budget survives them where an attempt cap gives up early. Default: None (attempt-based stopping).

• initial_backoff – Base wait in seconds before the first retry. Default: 2.0.

• max_backoff – Maximum wait in seconds between retries. The computed exponential wait is clamped to this value. Default: 64.0.

• multiplier – Exponential base applied to initial_backoff for each successive retry. Default: 2.0.

• jitter – Maximum random extra wait in seconds added to each exponential backoff (drawn uniformly from [0, jitter)). Helps avoid thundering-herd retries when many clients fail at once. Default: 0.0 (no jitter).

• unhealthy_backoff – Flat wait in seconds between 502 retries when the cluster reports unhealthy on /cluster_healthz. The cluster is likely upgrading/restarting, so a flat pause is preferable to an exponential ramp. Default: 90.0.

• retryable_status_codes – HTTP status codes that should trigger a retry. Default: {408, 429, 502, 503, 504}.

• expired_token_wait_seconds – How long to wait for a background refresher to replace a bearer token that has provably expired, when the credential is a callable and the instance answered 401. Only a token whose own exp claim has passed waits: a credential the instance refuses for any other reason still fails at once, so a wrong API key does not hang. The wait ends on the first token that is not itself expired, so a refresher minting from a skewed clock cannot end it early. This budget is separate from deadline_seconds: a request that exhausts both takes the sum of the two. Default: 0.0 (do not wait).

__init__(max_retries: int = 3, deadline_seconds: float | None = None, initial_backoff: float = 2.0, max_backoff: float = 64.0, multiplier: float = 2.0, jitter: float = 0.0, unhealthy_backoff: float = 90.0, expired_token_wait_seconds: float = 0.0, retryable_status_codes: ~typing.FrozenSet[int] = <factory>) → None

deadline_seconds: float | None = None

expired_token_wait_seconds: float = 0.0

initial_backoff: float = 2.0

jitter: float = 0.0

max_backoff: float = 64.0

max_retries: int = 3

multiplier: float = 2.0

retryable_status_codes: FrozenSet[int]

unhealthy_backoff: float = 90.0

feldera.rest.sql_table module

class feldera.rest.sql_table.SQLTable(name: str, fields: list[dict], case_sensitive: bool = False, materialized: bool = False)[source]

Bases: object

Represents a SQL table in Feldera

__init__(name: str, fields: list[dict], case_sensitive: bool = False, materialized: bool = False)[source]

classmethod from_dict(table_dict: dict)[source]

feldera.rest.sql_view module

class feldera.rest.sql_view.SQLView(name: str, fields: list[dict], case_sensitive: bool = False, materialized: bool = False)[source]

Bases: object

Represents a SQL view in Feldera

__init__(name: str, fields: list[dict], case_sensitive: bool = False, materialized: bool = False)[source]

classmethod from_dict(view_dict: dict)[source]