onestep

English | 简体中文

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:

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:

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")

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:

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:

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:

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:

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 for YAML policy.

What it does

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

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:

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:

pip install 'onestep[all]'

Configuration styles

Plain Python

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

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.

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

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.

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.

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

For files that cannot be inferred from imports, add build hints to pyproject.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/
• Official worker image — run YAML workers in Docker without packaging:

  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.
• Embed in a web app — recommended shape for FastAPI/Django in deploy/web-service-integration.md.

Control plane

onestep can push runtime telemetry (heartbeat, topology, metrics, events) to the onestep-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 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:

pip install 'onestep[control-plane]'

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:

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:

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.

Examples

Runnable examples live in example/. Highlights:

# 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 for the old-to-new API mapping, unsupported features, and rollout guidance.

More

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

License

MIT