> ## Documentation Index
> Fetch the complete documentation index at: https://docs.orbitra.atomo.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# client

# `orbitra.flows.client`

## Functions

### `get_flows_client`

```python theme={null}
get_flows_client(environment: str = 'prod') -> OrbitraFlowsClient
```

Get an authenticated OrbitraFlowsClient for the Orbitra Flows (Prefect) API.

**Args:**

* `environment`: The environment to get the API URL from. Defaults to "prod".

**Returns:**

* An authenticated OrbitraFlowsClient instance.

**Examples:**

```python theme={null}
from orbitra.flows.client import get_flows_client

client = get_flows_client()
```

## Classes

### `OrbitraFlowsClient`

A wrapper around SyncPrefectClient for interacting with Orbitra Flows.

**Methods:**

#### `api_url`

```python theme={null}
api_url(self) -> str
```

Get the API URL of the Orbitra Flows server.

#### `get_flow_run_logs_by_id`

```python theme={null}
get_flow_run_logs_by_id(self, flow_run_id: str | UUID, limit: Optional[int] = None) -> list[Log]
```

Get the logs emitted by a flow run, ordered oldest to newest.

**Args:**

* `flow_run_id`: The ID of the flow run to get logs for.
* `limit`: Maximum number of log records to return. Defaults to None (no limit).

**Returns:**

* list\[Log]: The flow run's logs sorted by timestamp ascending. Empty if the run has no
  logs or if no run exists with the given ID.

**Examples:**

```python theme={null}
from orbitra.flows.client import get_flows_client

client = get_flows_client()
logs = client.get_flow_run_logs_by_id("00000000-0000-0000-0000-000000000000")
for log in logs:
    print(log.timestamp, log.message)
```

#### `run_deployment`

```python theme={null}
run_deployment(*args: P.args, **kwargs: P.kwargs) -> str
```

Run a deployment by its decorated callable with type-safe parameters.

**Args:**

* `deployment`: The callable decorated with @orbitra\_deployment.
* `**kwargs`: Parameters forwarded to the flow run (type-checked against the flow signature).

**Returns:**

* The ID of the created flow run.

**Examples:**

```python theme={null}
from orbitra.flows.client import get_flows_client
from my_flows import my_flow

client = get_flows_client()
flow_run_id = client.run_deployment(my_flow, date="2024-01-01")
```

#### `run_deployment_by_name`

```python theme={null}
run_deployment_by_name(self, deployment_name: str, parameters: Optional[dict] = None) -> str
```

Run a Prefect deployment by name.

**Args:**

* `deployment_name`: The name of the deployment to run in the format \<FLOW\_NAME>/\<DEPLOYMENT\_NAME>.
* `parameters`: Parameters to pass to the flow run. Defaults to None.

**Returns:**

* The ID of the created flow run.

**Examples:**

```python theme={null}
from orbitra.flows.client import get_flows_client

client = get_flows_client()
flow_run_id = client.run_deployment_by_name("my-flow/my-deployment", parameters={"key": "value"})
```

#### `sync_prefect_client`

```python theme={null}
sync_prefect_client(self) -> SyncPrefectClient
```

The underlying
[SyncPrefectClient](https://docs.prefect.io/v3/api-ref/python/prefect-client-orchestration-__init__#syncprefectclient)
instance for direct Prefect API access.

#### `wait_for_flow_run`

```python theme={null}
wait_for_flow_run(self, flow_run_id: str | UUID, poll_interval: float = 5.0, timeout: Optional[float] = None, raise_on_flow_failure: bool = True) -> FlowRun
```

Poll a flow run until it reaches a terminal state.

**Args:**

* `flow_run_id`: The ID of the flow run to wait for.
* `poll_interval`: Seconds between polling attempts. Defaults to 5.0.
* `timeout`: Maximum seconds to wait before raising an error. Defaults to None (wait forever).
* `raise_on_flow_failure`: If True, raise FlowsError on flow failure. If False, return the FlowRun instead. Defaults to True.

**Returns:**

* The completed flow run object.

**Raises:**

* `FlowsError`: If the flow run fails, crashes, or is cancelled (only when raise\_on\_flow\_failure=True).

**Examples:**

```python theme={null}
from orbitra.flows.client import get_flows_client

client = get_flows_client()
flow_run_id = client.run_deployment_by_name("my-flow/my-deployment")
flow_run = client.wait_for_flow_run(flow_run_id, timeout=300)
```