import json

import logging

import pathlib

import time

import warnings

from decimal import Decimal

from typing import Any, Dict, Generator, List, Mapping, Optional

from urllib.parse import quote

import pyarrow as pa

import pyarrow.ipc

import requests

from feldera._long_operation_warning import LongOperationWarning

from feldera.enums import BootstrapPolicy, PipelineFieldSelector, PipelineStatus

from feldera.rest._helpers import determine_client_version

from feldera.rest._httprequests import HttpRequests

from feldera.rest.config import ApiKey, Config # noqa: F401 — re-exported

from feldera.rest.errors import FelderaAPIError, FelderaTimeoutError

from feldera.rest.feldera_config import FelderaConfig

from feldera.rest.pipeline import Pipeline

from feldera.tags import _normalize_tags

from feldera.rest.retry import RetryConfig

logger = logging.getLogger(__name__)

def _validate_no_none_keys_in_map(data):

def validate_no_none_keys(d: Dict[Any, Any]) -> None:

for k, v in d.items():

if isinstance(v, dict) and any(k is None for k in v.keys()):

raise ValueError("keys of SQL MAP objects cannot be NULL")

if isinstance(data, list):

for datum in data:

validate_no_none_keys(datum)

elif isinstance(data, dict):

validate_no_none_keys(data)

else:

return

def _prepare_boolean_input(value: bool) -> str:

return "true" if value else "false"

class FelderaClient:

"""

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

"""

def __init__(

self,

url: Optional[str] = None,

api_key: Optional[ApiKey] = None,

timeout: Optional[float] = None,

connection_timeout: Optional[float] = None,

requests_verify: Optional[bool | str] = None,

retry_config: Optional[RetryConfig] = None,

tenant: Optional[str] = None,

) -> None:

"""

Constructs a Feldera client.

:param 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"`.

:param 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).

:param 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).

:param 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).

:param 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`.

:param 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.

:param 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.

"""

self.config = Config(

url=url,

api_key=api_key,

timeout=timeout,

connection_timeout=connection_timeout,

requests_verify=requests_verify,

retry_config=retry_config,

tenant=tenant,

)

self.http = HttpRequests(self.config)

try:

client_version = determine_client_version()

server_config = self.get_config()

if client_version != server_config.version:

logger.warning(

f"Feldera client is on version {client_version} while server is at "

f"{server_config.version}. There could be incompatibilities."

)

except Exception as e:

logger.error(f"Failed to connect to Feldera API: {e}")

raise e

@staticmethod

def localhost(port: int = 8080) -> "FelderaClient":

"""

Create a FelderaClient that connects to the local Feldera instance

"""

return FelderaClient(f"http://127.0.0.1:{port}")

def get_pipeline(

self, pipeline_name: str, field_selector: PipelineFieldSelector

) -> Pipeline:

"""

Get a pipeline by name

:param pipeline_name: The name of the pipeline

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

"""

resp = self.http.get(

f"/pipelines/{pipeline_name}?selector={field_selector.value}"

)

return Pipeline.from_dict(resp)

def get_runtime_config(self, pipeline_name) -> Mapping[str, Any]:

"""

Get the runtime config of a pipeline by name

:param pipeline_name: The name of the pipeline

"""

resp: Mapping[str, Any] = self.http.get(f"/pipelines/{pipeline_name}")

runtime_config: Mapping[str, Any] | None = resp.get("runtime_config")

if runtime_config is None:

raise ValueError(f"Pipeline {pipeline_name} has no runtime config")

return runtime_config

def pipelines(

self, selector: PipelineFieldSelector = PipelineFieldSelector.STATUS

) -> list[Pipeline]:

"""

Get all pipelines

"""

resp = self.http.get(

path=f"/pipelines?selector={selector.value}",

)

return [Pipeline.from_dict(pipeline) for pipeline in resp]

def _wait_for_compilation(

self,

name: str,

expected_program_version: int | None = None,

timeout_s: float | None = None,

poll_interval_s: float = 1.0,

) -> Pipeline:

"""Wait for pipeline compilation -- internal use only."""

wait = ["Pending", "CompilingSql", "SqlCompiled", "CompilingRust"]

start_time = time.monotonic()

wait_for_compilation = LongOperationWarning(

logger,

lambda elapsed: f"still compiling {name}, waited {elapsed:.1f} seconds",

lambda elapsed: f"{name} finished compiling after {elapsed:.1f} seconds",

)

while True:

elapsed = time.monotonic() - start_time

if timeout_s is not None and elapsed > timeout_s:

raise TimeoutError(

f"Timed out waiting for pipeline '{name}' to compile "

f"(expected program_version >= {expected_program_version})"

)

p = self.get_pipeline(name, PipelineFieldSelector.STATUS)

status = p.program_status

if status == "Success":

wait_for_compilation.done()

if expected_program_version is None:

return self.get_pipeline(name, PipelineFieldSelector.ALL)

current_version = p.program_version or 0

if current_version == expected_program_version:

return self.get_pipeline(name, PipelineFieldSelector.ALL)

else:

raise RuntimeError(

f"program version ({current_version}) != expected program version ({expected_program_version})"

)

elif status not in wait:

p = self.get_pipeline(name, PipelineFieldSelector.ALL)

# error handling for SQL compilation errors

if status == "SqlError":

sql_errors = p.program_error["sql_compilation"]["messages"]

if sql_errors:

err_msg = f"Pipeline {name} failed to compile:\n"

for sql_error in sql_errors:

err_msg += (

f"{sql_error['error_type']}\n{sql_error['message']}\n"

)

err_msg += f"Code snippet:\n{sql_error['snippet']}"

raise RuntimeError(err_msg)

error_message = f"The program failed to compile: {status}\n"

rust_error = p.program_error.get("rust_compilation")

if rust_error is not None:

error_message += f"Rust Error: {rust_error}\n"

system_error = p.program_error.get("system_error")

if system_error is not None:

error_message += f"System Error: {system_error}"

raise RuntimeError(error_message)

wait_for_compilation.check()

time.sleep(poll_interval_s)

def __wait_for_pipeline_state(

self,

pipeline_name: str,

state: str,

timeout_s: Optional[float] = None,

start: bool = True,

poll_interval_s: float = 0.5,

):

start_time = time.monotonic()

wait_for_state = LongOperationWarning(

logger,

lambda elapsed: (

f"still waiting for {pipeline_name} to transition to "

f"'{state}', waited {elapsed:.1f} seconds"

),

lambda elapsed: (

f"{pipeline_name} transitioned to '{state}' after {elapsed:.1f} seconds"

),

)

while True:

if timeout_s is not None:

elapsed = time.monotonic() - start_time

if elapsed > timeout_s:

raise TimeoutError(

f"Timed out waiting for pipeline {pipeline_name} to "

f"transition to '{state}' state"

)

resp = self.get_pipeline(pipeline_name, PipelineFieldSelector.STATUS)

status = resp.deployment_status

if status.lower() == state.lower():

wait_for_state.done()

break

elif (

status == "Stopped"

and len(resp.deployment_error or {}) > 0

and resp.deployment_desired_status == "Stopped"

):

err_msg = "Unable to START the pipeline:\n" if start else ""

raise RuntimeError(

f"""{err_msg}Unable to transition the pipeline to '{state}'.

Reason: The pipeline is in a STOPPED state due to the following error:

{resp.deployment_error.get("message", "")}"""

)

wait_for_state.check()

time.sleep(poll_interval_s)

def __wait_for_pipeline_state_one_of(

self,

pipeline_name: str,

states: list[str],

timeout_s: float | None = None,

start: bool = True,

poll_interval_s: float = 0.5,

) -> PipelineStatus:

start_time = time.monotonic()

states = [state.lower() for state in states]

wait_for_states = LongOperationWarning(

logger,

lambda elapsed: (

f"still waiting for {pipeline_name} to transition to "

f"one of {states}, waited {elapsed:.1f} seconds"

),

lambda elapsed: f"{pipeline_name} transitioned after {elapsed:.1f} seconds",

)

while True:

if timeout_s is not None:

elapsed = time.monotonic() - start_time

if elapsed > timeout_s:

raise TimeoutError(

f"Timed out waiting for pipeline {pipeline_name} to"

f"transition to one of the states: {states}"

)

resp = self.get_pipeline(pipeline_name, PipelineFieldSelector.STATUS)

status = resp.deployment_status

if status.lower() in states:

wait_for_states.done()

return PipelineStatus.from_str(status)

elif (

status == "Stopped"

and len(resp.deployment_error or {}) > 0

and resp.deployment_desired_status == "Stopped"

):

err_msg = "Unable to START the pipeline:\n" if start else ""

raise RuntimeError(

f"""{err_msg}Unable to transition the pipeline to one of the states: {states}.

Reason: The pipeline is in a STOPPED state due to the following error:

{resp.deployment_error.get("message", "")}"""

)

wait_for_states.check()

time.sleep(poll_interval_s)

def create_pipeline(self, pipeline: Pipeline, wait: bool = True) -> Pipeline:

"""

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

:param pipeline: The pipeline to create

:param wait: Whether to wait for the pipeline to compile. True by default

"""

body = {

"name": pipeline.name,

"program_code": pipeline.program_code,

"udf_rust": pipeline.udf_rust,

"udf_toml": pipeline.udf_toml,

"program_config": pipeline.program_config,

"runtime_config": pipeline.runtime_config,

"description": pipeline.description or "",

"tags": _normalize_tags(pipeline.tags),

}

self.http.post(

path="/pipelines",

body=body,

)

if not wait:

return pipeline

return self._wait_for_compilation(pipeline.name)

def create_or_update_pipeline(

self, pipeline: Pipeline, wait: bool = True

) -> Pipeline:

"""

Create a pipeline if it doesn't exist or update a pipeline and wait for

it to compile

:param pipeline: The pipeline to create or update

:param wait: Whether to wait for the pipeline to compile. True by default

:return: The created or updated pipeline

"""

body = {

"name": pipeline.name,

"program_code": pipeline.program_code,

"udf_rust": pipeline.udf_rust,

"udf_toml": pipeline.udf_toml,

"program_config": pipeline.program_config,

"runtime_config": pipeline.runtime_config,

"description": pipeline.description or "",

"tags": _normalize_tags(pipeline.tags),

}

# Upsert by name: applying the same body twice yields the same

# pipeline, so a dropped connection is safe to retry.

self.http.put(

path=f"/pipelines/{pipeline.name}",

body=body,

idempotent=True,

)

if not wait:

return pipeline

return self._wait_for_compilation(pipeline.name)

def patch_pipeline(

self,

name: str,

sql: Optional[str] = None,

udf_rust: Optional[str] = None,

udf_toml: Optional[str] = None,

program_config: Optional[Mapping[str, Any]] = None,

runtime_config: Optional[Mapping[str, Any]] = None,

description: Optional[str] = None,

tags: Optional[List[str]] = None,

):

"""

Incrementally update pipeline

:param 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 :mod:`feldera.tags`).

"""

self.http.patch(

path=f"/pipelines/{name}",

body={

"program_code": sql,

"udf_rust": udf_rust,

"udf_toml": udf_toml,

"program_config": program_config,

"runtime_config": runtime_config,

"description": description,

"tags": _normalize_tags(tags) if tags is not None else None,

},

)

def testing_force_update_platform_version(self, name: str, platform_version: str):

self.http.post(

path=f"/pipelines/{name}/testing",

params={"set_platform_version": platform_version},

)

def update_pipeline_runtime(self, name: str):

"""

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

:param name: The name of the pipeline

"""

self.http.post(path=f"/pipelines/{name}/update_runtime")

def delete_pipeline(self, name: str):

"""

Deletes a pipeline by name

:param name: The name of the pipeline

"""

self.http.delete(

path=f"/pipelines/{name}",

idempotent=True,

)

def get_pipeline_stats(self, name: str) -> dict:

"""

Get the pipeline metrics and performance counters

:param name: The name of the pipeline

"""

resp = self.http.get(

path=f"/pipelines/{name}/stats",

)

return resp

def get_pipeline_metrics(self, name: str, format: str = "prometheus"):

"""

Get the pipeline circuit metrics

:param name: The name of the pipeline

:param format: The metrics format, "prometheus" (default) or "json".

:return: Prometheus exposition text as `str`, or a `dict` for "json".

"""

return self.http.get(

path=f"/pipelines/{name}/metrics",

params={"format": format},

)

def get_pipeline_logs(self, pipeline_name: str) -> Generator[str, None, None]:

"""

Get the pipeline logs

:param name: The name of the pipeline

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

"""

chunk: bytes

with self.http.get(

path=f"/pipelines/{pipeline_name}/logs",

stream=True,

) as resp:

for chunk in resp.iter_lines(chunk_size=50000000):

if chunk:

yield chunk.decode("utf-8")

def activate_pipeline(

self, pipeline_name: str, wait: bool = True, timeout_s: Optional[float] = None

) -> Optional[PipelineStatus]:

"""

:param pipeline_name: The name of the pipeline to activate

:param wait: Set True to wait for the pipeline to activate. True by

default

:param timeout_s: The amount of time in seconds to wait for the

pipeline to activate.

"""

# Desired-state setter: activating an already-activating pipeline is

# a no-op, so a dropped connection is safe to retry.

self.http.post(

path=f"/pipelines/{pipeline_name}/activate",

idempotent=True,

)

if not wait:

return None

return self.__wait_for_pipeline_state_one_of(

pipeline_name, ["running", "AwaitingApproval"], timeout_s

)

def _inner_start_pipeline(

self,

pipeline_name: str,

initial: str = "running",

bootstrap_policy: Optional[BootstrapPolicy] = None,

silent_bootstrap: bool = False,

concurrent_bootstrap: bool = False,

wait: bool = True,

timeout_s: Optional[float] = None,

dismiss_error: bool = True,

) -> Optional[PipelineStatus]:

"""

:param pipeline_name: The name of the pipeline to start

:param initial: The initial state to start the pipeline in. "running"

by default.

:param wait: Set True to wait for the pipeline to start. True by default

:param timeout_s: The amount of time in seconds to wait for the

pipeline to start.

:param dismiss_error: Set True to dismiss any deployment error before starting;

set False to make it fail in that case. True by default.

"""

start_params = {}

# Any error is dismissed separately in order to make sure we only dismiss any error

# BEFORE it was started, not after it has been started by this call

if dismiss_error:

self.http.post(

path=f"/pipelines/{pipeline_name}/dismiss_error", idempotent=True

)

start_params["dismiss_error"] = "false"

start_params["initial"] = initial

if bootstrap_policy is not None:

start_params["bootstrap_policy"] = bootstrap_policy.value

if silent_bootstrap:

start_params["silent_bootstrap"] = "true"

if concurrent_bootstrap:

start_params["concurrent_bootstrap"] = "true"

# Desired-state setters (start/pause/stop): repeating the request

# yields the same desired state, so a dropped connection is safe to

# retry.

self.http.post(

path=f"/pipelines/{pipeline_name}/start",

params=start_params,

idempotent=True,

)

if not wait:

return None

return self.__wait_for_pipeline_state_one_of(

pipeline_name,

[initial, "AwaitingApproval"],

timeout_s,

)

def start_pipeline(

self,

pipeline_name: str,

bootstrap_policy: Optional[BootstrapPolicy] = None,

silent_bootstrap: bool = False,

concurrent_bootstrap: bool = False,

wait: bool = True,

timeout_s: Optional[float] = None,

dismiss_error: bool = True,

) -> Optional[PipelineStatus]:

"""

:param pipeline_name: The name of the pipeline to start

:param wait: Set True to wait for the pipeline to start.

True by default

:param timeout_s: The amount of time in seconds to wait for the

pipeline to start.

:param dismiss_error: Set True to dismiss any deployment error before starting;

set False to make it fail in that case. True by default.

"""

return self._inner_start_pipeline(

pipeline_name,

"running",

bootstrap_policy,

silent_bootstrap,

concurrent_bootstrap,

wait,

timeout_s,

dismiss_error,

)

def start_pipeline_as_paused(

self,

pipeline_name: str,

bootstrap_policy: Optional[BootstrapPolicy] = None,

silent_bootstrap: bool = False,

concurrent_bootstrap: bool = False,

wait: bool = True,

timeout_s: float | None = None,

dismiss_error: bool = True,

) -> Optional[PipelineStatus]:

"""

:param pipeline_name: The name of the pipeline to start as paused.

:param wait: Set True to wait for the pipeline to start as pause.

True by default

:param timeout_s: The amount of time in seconds to wait for the

pipeline to start.

:param dismiss_error: Set True to dismiss any deployment error before starting;

set False to make it fail in that case. True by default.

"""

return self._inner_start_pipeline(

pipeline_name,

"paused",

bootstrap_policy,

silent_bootstrap,

concurrent_bootstrap,

wait,

timeout_s,

dismiss_error,

)

def start_pipeline_as_standby(

self,

pipeline_name: str,

bootstrap_policy: Optional[BootstrapPolicy] = None,

silent_bootstrap: bool = False,

concurrent_bootstrap: bool = False,

wait: bool = True,

timeout_s: Optional[float] = None,

dismiss_error: bool = True,

):

"""

:param pipeline_name: The name of the pipeline to start as standby.

:param wait: Set True to wait for the pipeline to start as standby.

True by default

:param timeout_s: The amount of time in seconds to wait for the

pipeline to start.

:param dismiss_error: Set True to dismiss any deployment error before starting;

set False to make it fail in that case. True by default.

"""

self._inner_start_pipeline(

pipeline_name,

"standby",

bootstrap_policy,

silent_bootstrap,

concurrent_bootstrap,

wait,

timeout_s,

dismiss_error,

)

def resume_pipeline(

self,

pipeline_name: str,

wait: bool = True,

timeout_s: Optional[float] = None,

):

"""

Resume a pipeline

:param pipeline_name: The name of the pipeline to stop

:param wait: Set True to wait for the pipeline to pause. True by default

:param timeout_s: The amount of time in seconds to wait for the pipeline

to pause.

"""

self.http.post(

path=f"/pipelines/{pipeline_name}/resume",

)

if not wait:

return

self.__wait_for_pipeline_state(pipeline_name, "running", timeout_s)

def pause_pipeline(

self,

pipeline_name: str,

wait: bool = True,

timeout_s: Optional[float] = None,

):

"""

Pause a pipeline

:param pipeline_name: The name of the pipeline to stop

:param error_message: The error message to show if the pipeline is in

STOPPED state due to a failure.

:param wait: Set True to wait for the pipeline to pause. True by default

:param timeout_s: The amount of time in seconds to wait for the pipeline

to pause.

"""

self.http.post(

path=f"/pipelines/{pipeline_name}/pause",

idempotent=True,

)

if not wait:

return

self.__wait_for_pipeline_state(pipeline_name, "paused", timeout_s)

def approve_pipeline(

self,

pipeline_name: str,

silent_bootstrap: bool = False,

concurrent_bootstrap: bool = False,

):

"""

Approve a pipeline awaiting approval to proceed with bootstrapping.

:param pipeline_name: The name of the pipeline to approve.

:param silent_bootstrap: Set True to bootstrap with output connectors

disabled, so no records are emitted during the bootstrap phase.

False by default.

:param 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.

"""

params = {}

if silent_bootstrap:

params["silent_bootstrap"] = "true"

if concurrent_bootstrap:

params["concurrent_bootstrap"] = "true"

self.http.post(

path=f"/pipelines/{pipeline_name}/approve",

params=params,

)

def stop_pipeline(

self,

pipeline_name: str,

force: bool,

wait: bool = True,

timeout_s: Optional[float] = None,

):

"""

Stop a pipeline

:param pipeline_name: The name of the pipeline to stop

:param force: Set True to immediately scale compute resources to zero.

Set False to automatically checkpoint before stopping.

:param wait: Set True to wait for the pipeline to stop. True by default

:param timeout_s: The amount of time in seconds to wait for the pipeline

to stop.

"""

params = {"force": str(force).lower()}

self.http.post(

path=f"/pipelines/{pipeline_name}/stop",

params=params,

idempotent=True,

)

if not wait:

return

start = time.monotonic()

wait_for_stop = LongOperationWarning(

logger,

lambda elapsed: (

f"still stopping {pipeline_name}, waited {elapsed:.1f} seconds"

),

lambda elapsed: f"{pipeline_name} stopped after {elapsed:.1f} seconds",

)

while True:

if timeout_s is not None and time.monotonic() - start > timeout_s:

raise FelderaTimeoutError(

f"timeout error: pipeline '{pipeline_name}' did not stop in {timeout_s} seconds"

)

status = self.get_pipeline(

pipeline_name, PipelineFieldSelector.STATUS

).deployment_status

if status == "Stopped":

wait_for_stop.done()

return

wait_for_stop.check()

time.sleep(0.1)

def dismiss_error_pipeline(

self,

pipeline_name: str,

):

"""

Dismisses the deployment error of a pipeline.

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

"""

self.http.post(path=f"/pipelines/{pipeline_name}/dismiss_error")

def clear_storage(

self,

pipeline_name: str,

wait: bool = True,

timeout_s: float | None = None,

poll_interval_s: float = 0.25,

):

"""

Clears the storage of the pipeline.

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

:param pipeline_name: Name of the pipeline.

:param wait: Set `True` to wait for the pipeline storage to become cleared,

or set `False` to immediately return. Default is `True`.

:param timeout_s: Timeout waiting for storage to become cleared.

`None` = no timeout is enforced (default). Not used if `wait=False`.

:param 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`.

"""

self.http.post(

path=f"/pipelines/{pipeline_name}/clear",

)

if not wait:

return

start = time.monotonic()

wait_for_clear = LongOperationWarning(

logger,

lambda elapsed: (

f"still clearing {pipeline_name}, waited {elapsed:.1f} seconds"

),

lambda elapsed: (

f"{pipeline_name} storage cleared after {elapsed:.1f} seconds"

),

)

while True:

if timeout_s is not None and time.monotonic() - start > timeout_s:

raise FelderaTimeoutError(

f"timeout error: pipeline '{pipeline_name}' did not clear storage in {timeout_s} seconds"

)

status = self.get_pipeline(

pipeline_name, PipelineFieldSelector.STATUS

).storage_status

if status == "Cleared":

wait_for_clear.done()

return

wait_for_clear.check()

time.sleep(poll_interval_s)

def start_transaction(self, pipeline_name: str) -> int:

"""

Start a new transaction.

Transaction ID.

:param pipeline_name: The name of the pipeline.

"""

resp = self.http.post(

path=f"/pipelines/{pipeline_name}/start_transaction",

)

return int(resp.get("transaction_id"))

def advance_clock(

self,

pipeline_name: str,

delta_ms: Optional[int] = None,

) -> Dict[str, int | str]:

"""

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.

:param pipeline_name: The name of the pipeline.

:param 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.

:return: 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.

"""

return self.http.post(

path=f"/pipelines/{pipeline_name}/clock/advance",

body={"delta_ms": delta_ms},

)

def pipeline_diff(

self,

pipeline_name: str,

program_code: Optional[str] = None,

runtime_version: Optional[str] = None,

) -> Dict[str, Any]:

"""

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

:param pipeline_name: The name of the pipeline.

:param program_code: New SQL program code to compare against. If

``None`` (the default), the pipeline's current program code is used.

:param 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.

:return: 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.

"""

body: Dict[str, Any] = {}

if program_code is not None:

body["program_code"] = program_code

if runtime_version is not None:

body["runtime_version"] = runtime_version

return self.http.post(

path=f"/pipelines/{pipeline_name}/diff",

body=body,

)

def validate_program(

self,