Skip to main content

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, 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+

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.

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
Emit results to a downstream sink any source doubles as a sink; MySQL table_sink; Kafka kafka_topic; 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
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

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

License

MIT

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

onestep-1.7.2.tar.gz (123.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

onestep-1.7.2-py3-none-any.whl (81.0 kB view details)

Uploaded Python 3

File details

Details for the file onestep-1.7.2.tar.gz.

File metadata

  • Download URL: onestep-1.7.2.tar.gz
  • Upload date:
  • Size: 123.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.14

File hashes

Hashes for onestep-1.7.2.tar.gz
Algorithm Hash digest
SHA256 392603b04ade809213f48f697fa048d8e510302bda77673ddd6ae4167edec020
MD5 f115b23a660c312ae72ff77d86ae2a02
BLAKE2b-256 1387a1d8cd54f63e62a42d1bc91d8df00d54ebf00228501d0f9a2392a4217543

See more details on using hashes here.

Provenance

The following attestation bundles were made for onestep-1.7.2.tar.gz:

Publisher: release.yml on mic1on/onestep

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file onestep-1.7.2-py3-none-any.whl.

File metadata

  • Download URL: onestep-1.7.2-py3-none-any.whl
  • Upload date:
  • Size: 81.0 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.14

File hashes

Hashes for onestep-1.7.2-py3-none-any.whl
Algorithm Hash digest
SHA256 bb7368b12fd54ff15a62e6d1effc70eff515f1efda6354c893e7274ac3832739
MD5 f0c485cd187107d81792451c16101e21
BLAKE2b-256 1a892dba82154eea6b233379da6eff5ad54a87002cee404e78aad2114cbcf620

See more details on using hashes here.

Provenance

The following attestation bundles were made for onestep-1.7.2-py3-none-any.whl:

Publisher: release.yml on mic1on/onestep

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Release history Release notifications | RSS feed

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page