Metadata-Version: 2.4
Name: onestep
Version: 1.9.0
Summary: Async task runtime for queue, polling, schedule, and webhook workloads.
Author-email: miclon <jcnd@163.com>
License: MIT
License-File: LICENSE
Keywords: async,queue,scheduler,tasks,webhook,worker
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.9
Provides-Extra: all
Requires-Dist: onestep-clickhouse>=0.1.0; extra == 'all'
Requires-Dist: onestep-control-plane>=0.1.1; extra == 'all'
Requires-Dist: onestep-elasticsearch>=0.1.0; extra == 'all'
Requires-Dist: onestep-kafka>=0.1.3; (python_version >= '3.10') and extra == 'all'
Requires-Dist: onestep-mongodb>=0.1.0; extra == 'all'
Requires-Dist: onestep-mq>=0.2.1; extra == 'all'
Requires-Dist: onestep-mysql>=0.3.3; extra == 'all'
Requires-Dist: onestep-postgres>=0.2.0; extra == 'all'
Requires-Dist: onestep-redis>=0.2.1; extra == 'all'
Requires-Dist: onestep-sqs>=0.2.2; extra == 'all'
Requires-Dist: pyyaml>=6.0; extra == 'all'
Provides-Extra: clickhouse
Requires-Dist: onestep-clickhouse>=0.1.0; extra == 'clickhouse'
Provides-Extra: control-plane
Requires-Dist: onestep-control-plane>=0.1.1; extra == 'control-plane'
Provides-Extra: dev
Requires-Dist: onestep-clickhouse>=0.1.0; extra == 'dev'
Requires-Dist: onestep-control-plane>=0.1.1; extra == 'dev'
Requires-Dist: onestep-elasticsearch>=0.1.0; extra == 'dev'
Requires-Dist: onestep-kafka>=0.1.3; (python_version >= '3.10') and extra == 'dev'
Requires-Dist: onestep-mongodb>=0.1.0; extra == 'dev'
Requires-Dist: onestep-mq>=0.2.1; extra == 'dev'
Requires-Dist: onestep-mysql>=0.3.3; extra == 'dev'
Requires-Dist: onestep-postgres>=0.2.0; extra == 'dev'
Requires-Dist: onestep-redis>=0.2.1; extra == 'dev'
Requires-Dist: onestep-sqs>=0.2.2; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: pyyaml>=6.0; extra == 'dev'
Provides-Extra: elasticsearch
Requires-Dist: onestep-elasticsearch>=0.1.0; extra == 'elasticsearch'
Provides-Extra: integration
Requires-Dist: onestep-clickhouse>=0.1.0; extra == 'integration'
Requires-Dist: onestep-elasticsearch>=0.1.0; extra == 'integration'
Requires-Dist: onestep-kafka>=0.1.3; (python_version >= '3.10') and extra == 'integration'
Requires-Dist: onestep-mongodb>=0.1.0; extra == 'integration'
Requires-Dist: onestep-mq>=0.2.1; extra == 'integration'
Requires-Dist: onestep-mysql>=0.3.3; extra == 'integration'
Requires-Dist: onestep-postgres>=0.2.0; extra == 'integration'
Requires-Dist: onestep-redis>=0.2.1; extra == 'integration'
Requires-Dist: onestep-sqs>=0.2.2; extra == 'integration'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'integration'
Requires-Dist: pytest>=8.0.0; extra == 'integration'
Requires-Dist: pyyaml>=6.0; extra == 'integration'
Provides-Extra: kafka
Requires-Dist: onestep-kafka>=0.1.3; (python_version >= '3.10') and extra == 'kafka'
Provides-Extra: mongodb
Requires-Dist: onestep-mongodb>=0.1.0; extra == 'mongodb'
Provides-Extra: mysql
Requires-Dist: onestep-mysql>=0.3.3; extra == 'mysql'
Provides-Extra: postgres
Requires-Dist: onestep-postgres>=0.2.0; extra == 'postgres'
Provides-Extra: rabbitmq
Requires-Dist: onestep-mq>=0.2.1; extra == 'rabbitmq'
Provides-Extra: redis
Requires-Dist: onestep-redis>=0.2.1; extra == 'redis'
Provides-Extra: sqs
Requires-Dist: onestep-sqs>=0.2.2; extra == 'sqs'
Provides-Extra: test
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'test'
Requires-Dist: pytest>=8.0.0; extra == 'test'
Requires-Dist: pyyaml>=6.0; extra == 'test'
Provides-Extra: yaml
Requires-Dist: pyyaml>=6.0; extra == 'yaml'
Description-Content-Type: text/markdown

# onestep

<div align=center><img src="https://onestep.code05.com/logo-3.svg" width="300"></div>
<div align=center>
<a href="https://pypi.org/project/onestep" target="_blank">
    <img src="https://img.shields.io/pypi/pyversions/onestep.svg" alt="Supported Python versions">
</a>
</div>

[English](README.md) | [简体中文](README.zh-CN.md)

<hr />

**onestep** is a small async task runtime for queue, polling, schedule, and
webhook workloads. You declare a task with a `source` and optional `sink`, and
the runtime takes care of fetching, concurrency, retries, dead-lettering, and
telemetry.

- **One decorator** turns any async function into a managed task
- **Pluggable connectors** for memory, MySQL, RabbitMQ, Redis, SQS, Kafka,
  Elasticsearch/OpenSearch, ClickHouse, MongoDB, and Feishu
- **Scheduling** via interval, cron, webhook, or DB-backed queues
- **Production-ready**: retries, dead-letter, timeouts, state stores, metrics,
  and an optional control-plane reporter
- **Two config styles**: plain Python, or declarative YAML
- Python 3.9+

## Quick start

Install:

```bash
pip install onestep
# optional extras:
pip install 'onestep[yaml]'          # YAML task definitions
pip install 'onestep[control-plane]' # push telemetry to onestep-control-plane
pip install 'onestep[kafka]'         # Kafka topic source/sink, Python 3.10+
pip install 'onestep[elasticsearch]' # Elasticsearch/OpenSearch bulk sink
pip install 'onestep[clickhouse]'    # ClickHouse table sink
pip install 'onestep[mongodb]'       # MongoDB polling, change streams, and sink
```

Define an app, then run it with the `onestep` CLI:

```python
from onestep import IntervalSource, OneStepApp

app = OneStepApp("billing-sync")


@app.task(source=IntervalSource.every(hours=1, immediate=True, overlap="skip"))
async def sync_billing(ctx, _):
    print("syncing billing data")
```

```bash
onestep run your_package.tasks:app
onestep check your_package.tasks:app   # validate the target before starting
```

### Logging

`onestep run` writes application logs, framework logs, and task lifecycle events
to stdout at INFO level by default. Application logger names do not need to use
the `onestep` namespace:

```python
import logging

logger = logging.getLogger("billing.kpi_sync")
```

Use `--log-level DEBUG` to include fetched, started, and sink-success details.
Use `--no-task-events` to disable the lifecycle logger installed by the CLI:

```bash
onestep run your_package.tasks:app --log-level DEBUG
onestep run your_package.tasks:app --no-task-events
```

An explicit `--log-level` overrides a level configured by the loaded target,
including YAML `app.logging.level`. Without the option, a target-configured level
is preserved; otherwise the CLI uses INFO. When the CLI installs the stdout
handler, that resolved level applies to arbitrary application logger names and
the `onestep` namespace. Existing logging handlers and custom
`StructuredEventLogger` instances are preserved; when a host has configured its
own handler, it also retains ownership of root logger levels.

Direct `app.run()` and `app.serve()` calls do not modify host logging or install
task event logging. Embedded applications retain full control of process logging.

## Local task diagnostics

Run exactly one task attempt from JSON, or replay a captured failure, without a
worker or control plane:

```bash
onestep task run your_package.tasks:app --task sync_billing --input input.json
onestep task replay your_package.tasks:app --task sync_billing --envelope captures/failure.json
onestep check your_package.tasks:app --connect
```

Diagnostics execute the real handler, task hooks, retry decision, and sink
routing. Sink I/O is suppressed by default; `--send` opens, sends to, and closes
the selected sinks. Handler and hook code may still perform external side
effects in either mode. `--timeout` defaults to 60 seconds and is enforced in a
spawned process, including for synchronously blocked code.

`delivery_action` is always a prediction because source `ack`/`retry`/`fail`
methods are synthetic. In dry-run, `would_dead_letter` means dead-lettering
would occur if the configured dead-letter sink publishes successfully. Use
`--send` to observe that result. A forced timeout during `--send` can leave a
partial external write and a later retry can duplicate it.

`check --connect` calls `open()` and `close()` only when both methods are
callable. State/cursor stores without that lifecycle are reported as
`not_probeable`; the command never calls `load()`, `save()`, or `delete()` as a
connectivity probe.

Opt-in failure capture makes production envelopes replayable:

```python
from onestep import FailureCaptureConfig, OneStepApp

app = OneStepApp(
    "billing-sync",
    failure_capture=FailureCaptureConfig(
        directory="captures",
        mode="terminal",
        redact_paths=("/body/customer/token",),
    ),
)
```

Capture files are versioned, private, atomically written, and reject lossy
serialization. `terminal` records only effective terminal failures; `all` also
records retryable attempts. Common values including datetime, UUID, bytes,
Decimal, enum, tuple/namedtuple, set, and frozenset round-trip losslessly.
Unsupported custom values produce an explicit capture error and no file rather
than a degraded record. See
[`docs/yaml-task-definition.md`](docs/yaml-task-definition.md) for YAML policy.

## What it does

| Capability | Where |
| --- | --- |
| **Fetch work** from a queue, schedule, webhook, or DB cursor | `MemoryQueue`, `IntervalSource`, `CronSource`, `WebhookSource`, MySQL `table_queue` / `incremental` / binlog, RabbitMQ `queue`, Redis `stream`, SQS `queue`, Kafka `kafka_topic`, MongoDB `mongodb_polling` / `mongodb_change_stream` |
| **Emit results** to a downstream sink | any source doubles as a sink; MySQL `table_sink`; Kafka `kafka_topic`; Elasticsearch/OpenSearch `elasticsearch_bulk_sink`; ClickHouse `clickhouse_table_sink`; MongoDB `mongodb_collection_sink`; HTTP `http_sink`; Feishu Bitable sink |
| **Schedule** recurring work | `IntervalSource.every(...)`, `CronSource(...)` with overlap control (`allow` / `skip` / `queue`) |
| **Ingest external events** | `WebhookSource` with bearer auth, shared listeners, body parsing |
| **Survive failures** | retry policies, `dead_letter` sink, per-task `timeout_s`, failure classification (`error` / `timeout` / `cancelled`) |
| **Track state** | `InMemoryStateStore`, MySQL state/cursor stores; `ctx.state` namespace per task |
| **Observe** | `@app.on_event` hooks, `InMemoryMetrics`, `StructuredEventLogger`, execution events |
| **Operate** | optional control-plane reporter with remote commands: `ping`, `shutdown`, `restart`, `drain`, `pause_task`, `resume_task`, `restart_task`, `sync_now` |

## Core concepts

The whole runtime is built on four ideas:

- **`OneStepApp`** — task registry and lifecycle manager
- **`Source`** — fetches data from a queue, schedule, webhook, or polling backend
- **`Sink`** — publishes processed results downstream
- **`Delivery`** — a single fetched item exposing `ack` / `retry` / `fail`

```python
from onestep import MemoryQueue, OneStepApp

app = OneStepApp("demo")
source = MemoryQueue("incoming")
sink = MemoryQueue("processed")


@app.task(source=source, emit=sink, concurrency=4)
async def double(ctx, item):
    return {"value": item["value"] * 2}


async def main():
    await source.publish({"value": 21})
    await app.serve()
```

## Connectors

Each backend ships as its own package so you only install what you use:

| Package | Provides | Install |
| --- | --- | --- |
| **core** | `MemoryQueue`, `IntervalSource`, `CronSource`, `WebhookSource`, `http_sink`, runtime | `pip install onestep` |
| **Control plane** | reporter telemetry and remote commands | `pip install 'onestep[control-plane]'` |
| **MySQL** | `table_queue`, `incremental`, binlog CDC, `table_sink`, state/cursor stores | `pip install onestep-mysql` |
| **PostgreSQL** | same primitives as MySQL, backed by PostgreSQL | `pip install onestep-postgres` |
| **RabbitMQ** | `queue` with exchange/routing-key binding and prefetch | `pip install onestep-mq` |
| **Redis** | `stream` with consumer groups, `XACK`, `XCLAIM`, `maxlen` | `pip install onestep-redis` |
| **SQS** | `queue` with batched deletes and heartbeat visibility | `pip install onestep-sqs` |
| **Kafka** | `kafka_topic` source/sink with manual offset commits | `pip install onestep-kafka` |
| **Feishu Bitable** | incremental source and upsert sink | `pip install onestep-feishu-bitable` |
| **Elasticsearch/OpenSearch** | `elasticsearch` connector and acknowledged `elasticsearch_bulk_sink` over the common REST bulk boundary | `pip install 'onestep[elasticsearch]'` (`onestep-elasticsearch`) |
| **ClickHouse** | `clickhouse` connector and acknowledged `clickhouse_table_sink` inserts into existing tables | `pip install 'onestep[clickhouse]'` (`onestep-clickhouse`) |
| **MongoDB** | `mongodb_polling`, raw `mongodb_change_stream` events, and `mongodb_collection_sink` insert/upsert | `pip install 'onestep[mongodb]'` (`onestep-mongodb`) |

The three database bulk sinks accept one mapping or a non-empty sequence of
mappings and await every backend chunk acknowledgement. onestep remains
at-least-once: a retry can repeat committed items or chunks, so use stable
document IDs, upsert keys, or a dedup-aware ClickHouse schema when duplicates
matter. A partial commit whose final write set is unknown is reported as
`UNCERTAIN` and is not automatically replayed.

The Elasticsearch plugin targets the common Elasticsearch/OpenSearch HTTP bulk
surface rather than either vendor's Python client. MongoDB polling and change
streams can use in-memory state for development, but production restart
guarantees require an explicit durable cursor store; change streams emit raw
events and default to `full_document: updateLookup`.

Or install everything at once:

```bash
pip install 'onestep[all]'
```

## Configuration styles

### Plain Python

Best for application code. Each connector is a class you instantiate:

```python
from onestep import OneStepApp
from onestep_redis import RedisConnector

app = OneStepApp("redis-demo")
redis = RedisConnector("redis://localhost:6379")
source = redis.stream("jobs", group="workers", batch_size=100)
out = redis.stream("processed")


@app.task(source=source, emit=out, concurrency=8)
async def process_job(ctx, item):
    return {"job": item["job"], "status": "done"}
```

### YAML

Best for deployment wiring. Keep business logic in Python; describe the
runtime — app, resources, hooks, tasks — declaratively.

```yaml
app:
  name: billing-sync

resources:
  tick:
    type: interval
    minutes: 5
    immediate: true

tasks:
  - name: sync_billing
    source: tick
    handler:
      ref: your_package.handlers.billing:sync_billing
```

```bash
onestep run worker.yaml
onestep check --strict worker.yaml   # schema validation, unknown-field detection
onestep init billing-sync            # scaffold a minimal YAML project
onestep build worker.yaml --out dist/worker.zip
```

The full YAML schema, resource types, conditional routing, and state binding
are covered in [`docs/yaml-task-definition.md`](docs/yaml-task-definition.md).

### Build a deployable worker package

`onestep build` packages a YAML worker project into a zip that a worker agent can
download and run. It validates the target first, collects the YAML entrypoint,
local Python modules referenced by handler, hook, and conditional routing refs,
dependency declaration files such as `pyproject.toml`, `requirements.txt`, and
`uv.lock`, packaging metadata such as README and license files, and writes an
`onestep-package.json` manifest into the zip.

```bash
onestep build worker.yaml --strict --out dist/worker.zip
```

For files that cannot be inferred from imports, add build hints to
`pyproject.toml`:

```toml
[tool.onestep.build]
entrypoint = "worker.yaml"
include = ["templates/**"]
exclude = ["templates/private/**"]
```

Use `--env-file .env` to provide local values for the pre-build check. `.env`
files are excluded from packages by default; deploy-time configuration should be
provided through the worker agent or control plane. The package manifest records
the entrypoint so compatible control-plane uploads can infer it automatically;
when uploading to an older control plane, pass the same entrypoint explicitly.
Use `--json` to emit the build report for automation.

## Deployment

- **systemd** — minimal unit + preflight check template in
  [`deploy/`](deploy/README.md)
- **Official worker image** — run YAML workers in Docker without packaging:
  ```bash
  docker run --rm \
    -e ONESTEP_TARGET=/workspace/worker.yaml \
    -v "$PWD:/workspace" \
    ghcr.io/mic1on/onestep-worker:1.7.2
  ```
  See [`deploy/worker-runtime-image.md`](deploy/worker-runtime-image.md).
- **Embed in a web app** — recommended shape for FastAPI/Django in
  [`deploy/web-service-integration.md`](deploy/web-service-integration.md).

## Control plane

`onestep` can push runtime telemetry (heartbeat, topology, metrics, events) to
the [`onestep-control-plane`](apps/control-plane) application over a single WebSocket and
accept remote commands — with no connector or task-code changes.

The host execution agent lives in [`apps/work-agent`](apps/work-agent) and is
published separately as `onestep-worker-agent`. It connects outbound to the
control plane and starts assigned workflow packages as local `onestep`
subprocesses.

Install the reporter plugin first:

```bash
pip install 'onestep[control-plane]'
```

```yaml
app:
  name: billing-sync

reporter: true
```

Required env: `ONESTEP_CONTROL_PLANE_URL`, `ONESTEP_CONTROL_PLANE_TOKEN`.
Optional service-level metadata can be reported with `reporter.service_description`
or `ONESTEP_SERVICE_DESCRIPTION` and shown in the control plane:

```yaml
reporter:
  service_description: Synchronizes billing data into the warehouse
```

Handlers can report low-cardinality custom counters and gauges through the same
reporter. The plane stores them and can expose them from its Prometheus
`/metrics` endpoint:

```python
async def sync_users(ctx, payload):
    success_count = 0
    failed_count = 0
    ...
    ctx.metrics.counter("rows_success").inc(success_count)
    ctx.metrics.counter("rows_failed").inc(failed_count)
    ctx.metrics.gauge("batch_size").set(success_count + failed_count)
```

For identity, multi-replica guidance, env vars, and a local demo, see
[`docs/stable-instance-identity.md`](docs/stable-instance-identity.md).

## Examples

Runnable examples live in [`example/`](example/README.md). Highlights:

```bash
# 5-second interval task
SYNC_INTERVAL_SECONDS=5 PYTHONPATH=src onestep run example.cli_app:app

# end-to-end: webhook -> queue -> worker -> dead-letter, with metrics + logs
PYTHONPATH=src python3 example/runtime_showcase.py
```

## Upgrading

`1.0.0` was a runtime rewrite. If you're coming from `0.5.x`, see
[`MIGRATION-0.5-to-1.0.0.md`](MIGRATION-0.5-to-1.0.0.md) for the old-to-new API
mapping, unsupported features, and rollout guidance.

## More

- [`docs/yaml-task-definition.md`](docs/yaml-task-definition.md) — YAML schema
- [`docs/core-reliability.md`](docs/core-reliability.md) — stable API,
  delivery semantics, plugin compatibility, and release checklist
- [`docs/framework-evolution-roadmap.md`](docs/framework-evolution-roadmap.md) —
  ordered framework milestones and exit gates
- [`docs/stable-instance-identity.md`](docs/stable-instance-identity.md) —
  reporter identity resolution
- [`docs/agent-ws-protocol.md`](docs/agent-ws-protocol.md) — agent WS protocol
- [`deploy/`](deploy/README.md) — deployment templates

## License

MIT
