Metadata-Version: 2.4
Name: ska-src-ef-broker-client
Version: 0.1.0
Summary: Client library for the SRCNet computing broker: submit jobs, watch them, fetch their logs.
Author: SKA Observatory
License-Expression: BSD-3-Clause
Project-URL: Homepage, https://gitlab.com/ska-telescope/src/deployments/skaosrc/ska-src-ef-computing-broker
Project-URL: Source, https://gitlab.com/ska-telescope/src/deployments/skaosrc/ska-src-ef-computing-broker
Keywords: ska,srcnet,computing-broker,workflow,htcondor
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Astronomy
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.31.0
Provides-Extra: test
Requires-Dist: pytest>=8.2.0; extra == "test"
Dynamic: license-file

# ska-src-ef-broker-client

Client library for the SRCNet computing broker: submit jobs, watch them, fetch
their logs.

One dependency (`requests`). Installing the broker service instead would pull
FastAPI, SQLAlchemy, Alembic, pika and the HTCondor bindings onto a machine that
only wants to submit a job.

```bash
pip install ska-src-ef-broker-client
```

## Run something

```python
from ska_src_ef_broker_client import Broker

broker = Broker("https://broker.example.org", token=my_access_token)

job = broker.run("echo hello", image="registry.example.org/tools/app:1")
job.wait()                 # blocks until terminal, returns the concise status
print(job.state)           # COMPLETE
print(job.logs())
```

You never name a site: you say what to run and which data it needs, and the
broker decides where that can happen.

### Follow the data

Passing Rucio data identifiers makes the broker resolve where they live and
restrict placement to the sites holding them:

```python
job = broker.run(
    "./analyse.sh /srcnet/input",
    image="registry.example.org/tools/analysis:2",
    dids=["SKA-Mid.integration:EB-1234.product-abcd"],
    input_url="dav://storm.example.org/sa/inputs",
)
```

`/srcnet/input` exists only when the job declares an input area — `run()` omits
`--input` otherwise rather than binding a path you never asked for.

### A Dask cluster

The broker starts a scheduler and workers as sibling tasks on one site and hands
your command the rendezvous file:

```python
broker.run_dask(
    "python analysis.py",         # your code is the cluster's client
    image="registry.example.org/tools/analysis-dask:1",
    workers=4,
).wait(timeout=3600)
```

**Your image runs all three roles**, so it has to contain Dask itself — the
scheduler and workers are `dask scheduler` / `dask worker` in *your* image, and
the workers execute your code, so they need your libraries too. One layer is
enough:

```dockerfile
FROM registry.example.org/tools/analysis:1
RUN pip install --no-cache-dir "dask[array,distributed]==2026.7.1"
```

An image without Dask fails fast with a message naming it. See
`docs/dask-image-contract.md` in the broker repository for the full contract and
a pinned base image to derive from.

```python
# inside analysis.py
import os
from dask.distributed import Client

client = Client(scheduler_file=os.environ["SKA_DASK_SCHEDULER_FILE"])
```

The cluster is torn down when your command exits.

### A rapthor pipeline

```python
broker.run_rapthor(
    data_dir="/srv/storage/site/sa/my-observation",
    image="registry.example.org/tools/rapthor:latest",
    cpu=8,
    threads=8,
)
```

## Watch and inspect

A `Job` caches nothing, so a job id is all you need later — including after a
notebook restart:

```python
job = broker.job("job-abc123")
job.state            # current state
job.status()         # concise view
job.logs()           # printable log tails
job.next_action()    # triage hint when a job looks stuck
job.cancel()

broker.jobs(limit=20)          # recent jobs
broker.is_ready()              # broker health
```

Defaults can live on the client instead of every call:

```python
broker = Broker(
    "https://broker.example.org",
    token=my_access_token,
    image="registry.example.org/tools/app:1",
    input_url="dav://storm.example.org/sa/inputs",
)
Broker.from_env()   # BROKER_URL / BROKER_TOKEN / BROKER_IMAGE / ...
```

## The transport layer

`BrokerClient` is one method per HTTP endpoint, each returning a
`requests.Response`, for callers who want the responses themselves:

```python
from ska_src_ef_broker_client import BrokerClient

client = BrokerClient("https://broker.example.org", bearer_token=token)
response = client.get_job("job-abc123")
print(response.json()["state"])
```

`Broker` is built on it — it fills in the submit payload, gives jobs an object
identity, and turns polling into `job.wait()`.

## Notes

* **A token is required.** Every call without one answers 401.
* **TLS**: for a private CA pass its bundle — `verify="/path/ca.crt"`. `verify=False`
  disables verification altogether, which exposes the bearer token these calls
  carry to anyone on the path; keep it out of anything but a throwaway probe.
* A worked example of all three workload classes lives in the broker repository
  at `demo/notebooks/broker_showcase.py` (a marimo notebook).
