Metadata-Version: 2.5
Name: onestep
Version: 1.12.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-cf-queues>=0.1.0; 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-redis>=0.2.1; extra == 'all'
Requires-Dist: onestep-sql[mysql,postgres,sqlite]>=0.2.0; 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: cloudflare
Requires-Dist: onestep-cf-queues>=0.1.0; extra == 'cloudflare'
Provides-Extra: control-plane
Requires-Dist: onestep-control-plane>=0.1.1; extra == 'control-plane'
Provides-Extra: dev
Requires-Dist: onestep-cf-queues>=0.1.0; 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-redis>=0.2.1; extra == 'dev'
Requires-Dist: onestep-sql[mysql,postgres,sqlite]>=0.2.0; 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-cf-queues>=0.1.0; 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-redis>=0.2.1; extra == 'integration'
Requires-Dist: onestep-sql[mysql,postgres,sqlite]>=0.2.0; 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: metrics
Provides-Extra: mongodb
Requires-Dist: onestep-mongodb>=0.1.0; extra == 'mongodb'
Provides-Extra: mysql
Requires-Dist: onestep-sql[mysql]>=0.1.0; extra == 'mysql'
Provides-Extra: postgres
Requires-Dist: onestep-sql[postgres]>=0.1.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: sql
Requires-Dist: onestep-sql[mysql,postgres,sqlite]>=0.2.0; extra == 'sql'
Provides-Extra: sqlite
Requires-Dist: onestep-sql[sqlite]>=0.2.0; extra == 'sqlite'
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
```

Or scaffold a ready-to-run project from a scenario template:

```bash
onestep init my-worker --template redis   # interval | webhook | redis | sql-cdc
cd my-worker
pip install -e .   # scaffold dependencies already pin the extras the template needs
onestep run worker.yaml
```

`onestep init --help` lists all templates; each generates a minimal `worker.yaml`
plus handler package and prints the `pip install` line it needs.

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.

#### Structured JSON logs

Use `--log-format json` to emit one JSON object per line instead of text —
ready for Loki, ELK, or any JSON log collector without a parsing pipeline:

```bash
onestep run your_package.tasks:app --log-format json
```

The format can also be pinned in YAML via `app.logging.format`, so deployments
(K8s/compose) get structured logs without changing the startup command:

```yaml
app:
  name: billing-sync
  logging:
    level: INFO
    format: json
```

An explicit `--log-format` flag overrides the YAML value; without either,
`text` is used.

Every line the CLI stdout handler emits becomes a JSON object with `ts`,
`level`, `logger`, and `message` keys. Task lifecycle events logged by the
built-in `StructuredEventLogger` additionally promote their structured fields
to the top level, so platforms can index them directly:

```json
{"ts": "2026-09-03T01:14:01.808690+00:00", "level": "INFO", "logger": "onestep.events", "message": "task succeeded", "event_kind": "succeeded", "app_name": "billing", "task_name": "sync", "source_name": "queue.in", "attempts": 1, "duration_s": 0.0004, "failure_kind": null, "failure_message": null}
```

Other log records keep their `extra={...}` attributes under a nested `extra`
key. Unserializable values fall back to their `repr` so logging never raises.
The formatter is also usable directly for embedded setups:

```python
import logging

from onestep import JsonLogFormatter

handler = logging.StreamHandler()
JsonLogFormatter.attach(handler)
logging.getLogger().addHandler(handler)
```

The default `text` behavior is unchanged.

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.

## Render the worker topology

`onestep render` prints the topology of any Python or YAML target as a
[Mermaid](https://mermaid.js.org) flowchart, ready to paste into GitHub,
Notion, or Obsidian:

```bash
onestep render worker.yaml
```

```text
graph LR
  %% app: billing-sync
  n0["extract_entities<br/>concurrency=4 · retry=NoRetry · timeout=300s"]
  n1["sqs-orders<br/>MemoryQueue"]
  n2["mysql.meta_sink<br/>MemoryQueue"]
  n1 --> n0
  n0 -->|"emit"| n2
```

Each task node lists its concurrency, retry policy, and timeout. Edges are
labeled `emit` (with the transform ref when a binding sets one), `when`/`otherwise`
for conditional routes, and dashed `dead_letter` edges. Resources shared by
multiple tasks appear once, so chained topologies are drawn as connected graphs.

## 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`, Cloudflare `cf_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-sql[mysql]'` (`onestep-mysql` shim available) |
| **PostgreSQL** | same primitives as MySQL, backed by PostgreSQL | `pip install 'onestep-sql[postgres]'` (`onestep-postgres` shim available) |
| **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, plus an `sns_topic` fan-out sink | `pip install onestep-sqs` |
| **Cloudflare Queues** | `cf_queue` HTTP pull-consumer source/sink (official `cloudflare` SDK) with batched lease ack/retry | `pip install 'onestep[cloudflare]'` (`onestep-cf-queues`) |
| **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`.

### Which connector should I use?

| You want to... | Use | Install |
| --- | --- | --- |
| Run a task every N seconds / on a cron schedule | `interval` / `cron` (core) | `pip install 'onestep[yaml]'` |
| Receive HTTP webhooks | `webhook` (core) | `pip install 'onestep[yaml]'` |
| Consume a Redis Stream | `redis_stream` | `pip install 'onestep[redis,yaml]'` |
| Consume a RabbitMQ queue | `rabbitmq_queue` | `pip install 'onestep[rabbitmq,yaml]'` |
| Consume an AWS SQS queue | `sqs_queue` | `pip install 'onestep[sqs,yaml]'` |
| Consume a Cloudflare Queue | `cf_queue` | `pip install 'onestep[cloudflare,yaml]'` |
| Poll new/changed MySQL rows | `mysql_incremental` | `pip install 'onestep[mysql,yaml]'` |
| Stream MySQL binlog changes (CDC) | `mysql_binlog` | `pip install 'onestep[mysql,yaml]'` |
| Use a MySQL table as a work queue | `mysql_table_queue` | `pip install 'onestep[mysql,yaml]'` |
| Poll new/changed PostgreSQL rows | `postgres_incremental` | `pip install 'onestep[postgres,yaml]'` |
| Claim jobs from PostgreSQL | `postgres_execution_source` | `pip install 'onestep[postgres,yaml]'` |
| Consume a Kafka topic | `kafka_topic` | `pip install 'onestep[kafka,yaml]'` (Python 3.10+) |
| Poll MongoDB / watch change streams | `mongodb_polling` / `mongodb_change_stream` | `pip install 'onestep[mongodb,yaml]'` |
| Read SaaS tables (Feishu Bitable) | `feishu_bitable_incremental` | `pip install onestep-feishu-bitable` |
| Write to MySQL / PostgreSQL | `mysql_table_sink` / `postgres_table_sink` | `pip install 'onestep[mysql,yaml]'` / `pip install 'onestep[postgres,yaml]'` |
| Write to MongoDB | `mongodb_collection_sink` | `pip install 'onestep[mongodb,yaml]'` |
| Write to Elasticsearch/OpenSearch | `elasticsearch_bulk_sink` | `pip install 'onestep[elasticsearch,yaml]'` |
| Write to ClickHouse | `clickhouse_table_sink` | `pip install 'onestep[clickhouse,yaml]'` |
| Call an HTTP endpoint per item | `http_sink` (core) | `pip install 'onestep[yaml]'` |

Prefer `interval` for prototypes and scheduled jobs, a queue connector
(Redis/RabbitMQ/SQS/Cloudflare/Kafka) when work arrives as events or needs
competing consumers, and the CDC/incremental sources when the source of truth
is a database table. `onestep init --template {interval,webhook,redis,sql-cdc}`
scaffolds the four most common setups as ready-to-run projects.

### Minimal YAML per connector

Each snippet shows only the `resources:` block; pair it with a task that binds
your handler:

```yaml
tasks:
  - name: run
    source: <source-resource>
    emit: [<sink-resource>]        # optional
    handler:
      ref: your_pkg.tasks:run
```

Core (built-in):

```yaml
resources:
  tick:
    type: interval
    seconds: 60
    immediate: true
  nightly:
    type: cron
    expression: "0 3 * * *"
  intake:
    type: webhook
    path: /hooks/in
    methods: [POST]
  notify:
    type: http_sink
    url: https://example.invalid/hook
```

Redis Streams (`pip install 'onestep[redis,yaml]'`):

```yaml
resources:
  redis:
    type: redis
    url: redis://localhost:6379
  jobs:
    type: redis_stream
    connector: redis
    stream: jobs
    group: workers
    create_group: true
```

RabbitMQ (`pip install 'onestep[rabbitmq,yaml]'`):

```yaml
resources:
  rmq:
    type: rabbitmq
    url: amqp://guest:guest@localhost/
  jobs:
    type: rabbitmq_queue
    connector: rmq
    queue: incoming_jobs
    prefetch: 50
```

AWS SQS (`pip install 'onestep[sqs,yaml]'`):

```yaml
resources:
  sqs:
    type: sqs
    region_name: us-east-1
  jobs:
    type: sqs_queue
    connector: sqs
    url: https://sqs.us-east-1.amazonaws.com/123456789012/jobs
```

Cloudflare Queues (`pip install 'onestep[cloudflare,yaml]'`):

```yaml
resources:
  cf:
    type: cf_queues
    account_id: your-account-id
    api_token: your-api-token
  jobs:
    type: cf_queue
    connector: cf
    queue_id: your-queue-id
```

MySQL (`pip install 'onestep[mysql,yaml]'`) — incremental polling, binlog CDC,
and a table sink share one connector; CDC needs a cursor store to resume:

```yaml
resources:
  db:
    type: mysql
    dsn: mysql+pymysql://user:password@localhost:3306/app
  cursor:
    type: mysql_cursor_store
    connector: db
  changed_rows:
    type: mysql_incremental
    connector: db
    table: orders
    key: id
    cursor: [updated_at, id]
    state: cursor
  cdc:
    type: mysql_binlog
    connector: db
    server_id: 18491
    schemas: [app]
    tables: [orders]
    state: cursor
    state_key: orders-cdc
  processed:
    type: mysql_table_sink
    connector: db
    table: processed_orders
    mode: upsert
    keys: [id]
```

PostgreSQL (`pip install 'onestep[postgres,yaml]'`):

```yaml
resources:
  db:
    type: postgres
    dsn: postgresql+psycopg://user:password@localhost:5432/app
  cursor:
    type: postgres_cursor_store
    connector: db
  changed_rows:
    type: postgres_incremental
    connector: db
    table: orders
    key: id
    cursor: [updated_at, id]
    state: cursor
  jobs:
    type: postgres_execution_source
    connector: db
    namespace: my-worker
    task_names: [run_report]
  processed:
    type: postgres_table_sink
    connector: db
    table: processed_orders
```

Kafka (`pip install 'onestep[kafka,yaml]'`, Python 3.10+):

```yaml
resources:
  kafka:
    type: kafka
    bootstrap_servers: localhost:9092
  orders:
    type: kafka_topic
    connector: kafka
    topic: orders.events
    group_id: orders-workers
```

MongoDB (`pip install 'onestep[mongodb,yaml]'`) — `state` is omitted here so
polling/change streams use in-memory cursors for development; wire a durable
cursor store (e.g. `postgres_cursor_store`) via `state:` for restart guarantees:

```yaml
resources:
  mongo:
    type: mongodb
    uri: mongodb://localhost:27017
    database: app
  changes:
    type: mongodb_change_stream
    connector: mongo
    collection: events
  archive:
    type: mongodb_collection_sink
    connector: mongo
    collection: archive
    mode: upsert
    keys: [event_id]
```

Elasticsearch/OpenSearch (`pip install 'onestep[elasticsearch,yaml]'`):

```yaml
resources:
  search:
    type: elasticsearch
    hosts: ["https://localhost:9200"]
  indexed:
    type: elasticsearch_bulk_sink
    connector: search
    index: events
    operation: index
```

ClickHouse (`pip install 'onestep[clickhouse,yaml]'`):

```yaml
resources:
  db:
    type: clickhouse
    dsn: http://localhost:8123/default
  events:
    type: clickhouse_table_sink
    connector: db
    table: events
```

Feishu Bitable (`pip install onestep-feishu-bitable`):

```yaml
resources:
  bitable:
    type: feishu_bitable
    app_id: your-app-id
    app_secret: your-app-secret
  rows:
    type: feishu_bitable_incremental
    connector: bitable
    app_token: your-app-token
    table_id: tblxxx
    cursor_field: updated_at
```

Or install everything at once:

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

> **MySQL/PostgreSQL consolidation (issue #133):** `onestep-sql` is now the
> canonical distribution for both MySQL and PostgreSQL. New deployments should
> install `onestep-sql[mysql]` / `onestep-sql[postgres]` (or `onestep[mysql]` /
> `onestep[postgres]`). The legacy `onestep-mysql` / `onestep-postgres` packages
> remain available as thin forwarding shims — existing `pip install
> onestep-mysql` and `from onestep_mysql import ...` imports keep working
> unchanged. All 14 YAML resource type names are unchanged. See
> [the migration guide](docs/guide/migrate-to-onestep-sql.md) for details.

## 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 render worker.yaml           # worker topology as a Mermaid diagram
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
