Metadata-Version: 2.5
Name: bazhuayu-dataapp
Version: 0.0.2
Summary: Developer SDK for building Bazhuayu DataHub Data Apps: read input, push records, handle aborts
Project-URL: Homepage, https://www.bazhuayu.com
Author: Bazhuayu DataHub Team
License: MIT
License-File: LICENSE
Keywords: bazhuayu,data-app,datahub,sdk
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
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: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: jsonschema>=4.0
Description-Content-Type: text/markdown

# bazhuayu-dataapp

The developer SDK for building Bazhuayu DataHub Data Apps (`bazhuayu-dataapp`, imported as
`bazhuayu_dataapp`). Your `app.py` is the program; the SDK is a library you call inside
`async with DataApp:` to read the run input, push records and react to aborts.

To discover, run and consume Data Apps as a caller, use `bazhuayu-client` instead.

## Installation

```bash
uv add bazhuayu-dataapp
```

Start from the project template (`GET /v1/templates/python.zip`): it ships the `.dataapp/`
contract, `app.py`, `Dockerfile` and `AGENTS.md`.

## Usage

```python
import asyncio

from bazhuayu_dataapp import DataApp, Event


async def main() -> None:
    async with DataApp:                        # enter: fetch the run input
        inp = await DataApp.get_input()

        async def on_abort(event) -> None:     # cancel / timeout / record cap
            await DataApp.exit()
        DataApp.on(Event.ABORTING, on_abort)

        api_key = DataApp.get_env("example_api_key")   # alias from .dataapp/runtime.json
        for i, keyword in enumerate(inp["keywords"], 1):
            DataApp.log.info("processing %s", keyword)
            await DataApp.push_data({"id": keyword, "title": f"Result for {keyword}"})
            await DataApp.set_progress(i, len(inp["keywords"]))
                                               # leave: report the final state


if __name__ == "__main__":
    asyncio.run(main())
```

Run it with `uv run python app.py`, both locally and on the platform.

## API

| API | Description |
|---|---|
| `async with DataApp:` | Starts the run. Leaving normally reports `succeeded`; an uncaught exception reports `failed` (or `partially_succeeded` if the platform has accepted records). |
| `await DataApp.get_input()` | The run input, with top-level `default`s from `input_schema.json` applied. |
| `await DataApp.push_data(record_or_list)` | Pushes records. Each record must be JSON-serializable (convert `datetime`, `Decimal`, `bytes`, NaN first) and match `output_schema.json`. The SDK batches writes and retries transient failures in the background; if the platform keeps refusing writes, `push_data` waits instead of buffering without limit. A batch the platform rejects outright fails the run with `push-rejected`. |
| `DataApp.get_env(alias, default=None)` | Reads an env entry declared in `runtime.json` `env`. Treat values as secrets. |
| `await DataApp.set_progress(done, total, status_text=None, *, upstream_ref=None)` | Progress and a human-readable status. Throttled; only the latest value is sent. `upstream_ref` is the upstream system's own task id, shown on the run for tracing. It is kept until a progress report succeeds, even if later progress values replace the rest. `status_text` longer than 500 characters is truncated. |
| `DataApp.on(Event.ABORTING, handler)` | Called once on cancel, timeout, record cap or SIGTERM/SIGINT, with `{"reason": ...}`. Without a handler the SDK calls `exit()` for you. On timeout, if the run is still going 15 seconds after your handler returns, the SDK ends it (final state `run-timeout`); cancel stays cooperative. |
| `await DataApp.exit()` | Flushes, reports the final state and ends the `async with` block. |
| `await DataApp.fail(message, code=..., retryable=False)` | Ends the run with a structured error. |
| `DataApp.log` | A `logging.Logger` writing to stderr; env values are masked. |
| `DataApp.is_at_home` | `True` on the platform, `False` locally. |

Custom billing quantities are derived by the platform from pushed records
(`pricing.json` `events[].qty_from`); there is no manual charge call.

## Local mode

Without `DATAHUB_SERVER_URL` and `DATAHUB_WORKER_TOKEN` the SDK runs locally:

- input is read from `storage/input.json` and validated against `.dataapp/input_schema.json`;
- records are written to `storage/dataset/records.jsonl` (cleared on each run) and checked
  against `.dataapp/output_schema.json` (warnings only);
- `get_env` reads the process environment, then `.env` in the project root;
- progress and the final state are logged; Ctrl-C triggers `ABORTING`, a second Ctrl-C
  stops immediately; `fail()` exits with status 1.

## Platform mode

The platform runs your image as is (its own `ENTRYPOINT` / `CMD`, e.g. `python app.py`) and
injects `DATAHUB_SERVER_URL` and `DATAHUB_WORKER_TOKEN` (a credential scoped to this one run).
On entering `async with DataApp:` the SDK claims the run, then fetches the env entries
declared in `runtime.json` if there are any. The run time limit comes from
`execution.timeout_seconds`. Every request carries the runtime identity header
`X-DataHub-Runtime: lang=python; sdk=<version>; protocol=1`.

If the claim returns nothing to run (the run was cancelled before it started, or this SDK version is no
longer supported and the platform has already failed the run), the process exits with status 0
and the `async with` body never runs.

Exit codes only matter if the process exits before it claims its run. The platform reads them as follows:

| Code | Meaning |
|---|---|
| `0` | Nothing to run, or the claimed run already reached a final state |
| `2` | Only one of `DATAHUB_SERVER_URL` / `DATAHUB_WORKER_TOKEN` is set (a wrapper dropped the environment) |
| `75` | The platform could not be reached or rejected the credential; the platform retries the run |

Do not exit with `75` from your own code: a pre-claim exit with that code is treated as a platform
failure and retried.

## Logs

On the platform, everything your app writes to stdout and stderr (`DataApp.log`, `print`
and uncaught errors) becomes the run's log. The person who started the run can read it, for
test runs and production runs alike; the app owner cannot read the logs of runs started by
others. The output of your image build is the build log, visible to you as the owner.

Before a log is stored, the platform masks the env values handed to the run, inputs marked
`sensitive`, platform credentials and common credential patterns. Anything else you print is
shown as is, so do not print personal data or secrets you obtained in other ways. Run logs
are kept for 30 days. A run keeps at most 2 MiB of log (the beginning and the end of each
attempt), and lines longer than 16 KiB are cut.

## License

MIT
