onestep 1.4.5

Async task runtime for queue, polling, schedule, and webhook workloads.

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, Feishu
• Scheduling via interval, cron, webhook, or DB-backed queues
• Production-ready: retries, dead-letter, timeouts, state stores, metrics, and a 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

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

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
Emit results to a downstream sink	any source doubles as a sink; MySQL table_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	control-plane reporter with remote commands: ping, shutdown, restart, drain, pause_task, resume_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

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, reporter	pip install onestep
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
Feishu Bitable	incremental source and upsert sink	pip install onestep-feishu-bitable

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

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

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.2.7
  

  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 onestep-control-plane over a single WebSocket and accept remote commands — with no connector or task-code changes.

app:
  name: billing-sync

reporter: true

Required env: ONESTEP_CONTROL_PLANE_URL, ONESTEP_CONTROL_PLANE_TOKEN.

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/stable-instance-identity.md — reporter identity resolution
• docs/agent-ws-protocol.md — agent WS protocol
• deploy/ — deployment templates

License

MIT