onestep

onestep is a small async task runtime centered on four concepts:

OneStepApp: task registry and lifecycle manager
Source: fetch data from a queue or polling backend
Sink: publish processed data
Delivery: a single fetched item with ack/retry/fail

The V1 stable surface includes:

MemoryQueue
onestep-mysql: MySQLConnector.table_queue(...), binlog and incremental sources, table sinks, state stores, and cursor stores
onestep-mq: RabbitMQConnector.queue(...)
onestep-redis: RedisConnector.stream(...)
onestep-sqs: SQSConnector.queue(...)
IntervalSource.every(...)
CronSource(...)
WebhookSource(...)

Install

Core package:

pip install onestep

Common extras:

pip install 'onestep[yaml]'
pip install onestep-mysql
pip install onestep-mq
pip install onestep-redis
pip install onestep-sqs
pip install 'onestep[control-plane]'
pip install 'onestep[all]'

Connector plugins:

Feishu Bitable: install onestep-feishu-bitable
MySQL: install onestep-mysql
RabbitMQ: install onestep-mq
Redis: install onestep-redis
SQS: install onestep-sqs

From a source checkout:

pip install -e .
pip install -e '.[dev]'
pip install -e '.[integration]'

Upgrading from 0.5.x

1.0.0 is a runtime rewrite. Projects built on the legacy step and *Broker APIs should treat the upgrade as a migration, not a drop-in package bump.

See MIGRATION-0.5-to-1.0.0.md for:

old-to-new API mapping
unsupported legacy features
a minimal before/after example
rollout guidance for existing projects

CLI

The deployment entrypoint is the onestep CLI.

Recommended module shape:

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

Run the app:

onestep run your_package.tasks:app

The short form is also supported:

onestep your_package.tasks:app

Check the target before starting it:

onestep check your_package.tasks:app

You can also point the CLI at a zero-argument factory:

onestep check your_package.tasks:build_app

Use JSON output when you want the check result in CI or deployment scripts:

onestep check --json your_package.tasks:app

You can also load a YAML app definition. Start with the smallest working shape and add fields only when needed. resources is the preferred top-level section for named runtime objects, while handler.ref and hooks.*.ref point to Python callables:

app:
  name: billing-sync
  logging:
    level: DEBUG

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

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

Check or run the YAML target the same way:

onestep check worker.yaml
onestep run worker.yaml

Scaffold a minimal standalone YAML project when starting from scratch:

onestep init billing-sync

That generates:

pyproject.toml
worker.yaml
src/<package>/tasks/
src/<package>/transforms/

The generated project stays intentionally small. worker.yaml only defines runtime wiring; business logic still lives in Python. Add hooks.py later only when you actually wire hooks in YAML.

Use strict checking when you want schema validation for YAML targets, including unknown-field detection and contract checks for reporter, hooks, tasks, and resource specs:

onestep check --strict worker.yaml

app.logging.level is optional and only controls the onestep logger namespace. It does not configure handlers or formatters, but setting it to DEBUG makes framework-level sink success logs visible.

YAML resources can reference other resources by name, for example rabbitmq_queue.connector: rmq or mysql_incremental.state: cursor_store.

YAML emit entries can also route handler results with a Python predicate. The YAML stays declarative: when names a callable, while the condition logic stays in Python.

emit:
  - audit_sink
  - when:
      ref: your_package.routing:is_active_user
      params:
        status_field: status
    then: active_user_sink
    otherwise: inactive_user_sink

The predicate may accept ctx, payload, and result; then and otherwise can each be a sink name or list of sink names. If otherwise is omitted and the predicate is falsy, that route emits to no sink.

Task handlers and hooks can read:

ctx.config for app-level config
ctx.task_config for task-level config
ctx.resources for named runtime objects

App hooks can read app.resources for the same named resource registry.

For a progressive YAML guide from minimal task to fully wired app, including strict-check guidance, see docs/yaml-task-definition.md.

For a standalone project-style example with worker.yaml plus tasks.py / transforms.py / hooks.py, see example/yaml_project/.

Built-in YAML resource types:

memory
interval
cron
webhook
http_sink

Plugin YAML resource types:

onestep-mysql: mysql, mysql_state_store, mysql_cursor_store, mysql_table_queue, mysql_incremental, mysql_binlog, mysql_table_sink
onestep-mq: rabbitmq, rabbitmq_queue
onestep-redis: redis, redis_stream
onestep-sqs: sqs, sqs_queue
onestep-feishu-bitable: feishu_bitable, feishu_bitable_incremental, feishu_bitable_table_sink

Install the corresponding plugin package in the worker environment before using plugin resource types in YAML.

http_sink sends task results as JSON by default. Configure body only when the outbound payload should be reshaped. url, headers, params, and configured body values can reference body, payload, meta, and attempts with {{ ... }} variables:

resources:
  notify:
    type: http_sink
    url: "https://api.example.com/orders/{{ body.order_id }}"
    method: POST
    headers:
      X-Trace-Id: "{{ meta.trace_id }}"
    params:
      attempt: "{{ attempts }}"
    body:
      id: "{{ body.order_id }}"
      status: "{{ body.status }}"

YAML apps can also bind app-level state explicitly:

app:
  name: billing-sync
  state: app_state

connectors:
  db:
    type: mysql
    dsn: mysql+pymysql://root:root@localhost:3306/app
  app_state:
    type: mysql_state_store
    connector: db
    table: onestep_state

The named state resource must support load/save/delete; mysql_state_store and mysql_cursor_store both work when onestep-mysql is installed.

Legacy connectors, sources, and sinks sections are still supported and merged into the same internal resource registry. For the full YAML schema and a larger task-definition example, see docs/yaml-task-definition.md.

Runnable examples live in:

example/cli_app.py
example/cli_app.yaml

Run it locally with:

PYTHONPATH=src onestep check example.cli_app:app
onestep check example/cli_app.yaml
SYNC_INTERVAL_SECONDS=5 PYTHONPATH=src onestep run example.cli_app:app

systemd

A complete deployment template lives in:

deploy/README.md
deploy/systemd/onestep-app.service
deploy/env/onestep-app.env.example
deploy/bin/onestep-preflight.sh

The example uses:

/etc/onestep/onestep-app.env for deployment variables
ExecStartPre to run a startup check
ExecStart to launch onestep run

Install it with:

sudo mkdir -p /etc/onestep
sudo cp deploy/env/onestep-app.env.example /etc/onestep/onestep-app.env
sudo cp deploy/systemd/onestep-app.service /etc/systemd/system/onestep-app.service
sudo systemctl daemon-reload
sudo systemctl enable --now onestep-app

Check status and logs:

sudo systemctl status onestep-app
sudo journalctl -u onestep-app -f

See deploy/README.md for the expected directory layout and the env vars you need to adjust first. The deploy template prepends APP_CWD to PYTHONPATH so module targets defined inside the repo can be imported by the onestep console script.

Official worker image

onestep also ships an official worker runtime image for YAML-oriented workers.

Mounted workspace usage:

docker run --rm \
  -e ONESTEP_TARGET=/workspace/worker.yaml \
  -v "$PWD:/workspace" \
  ghcr.io/mic1on/onestep-worker:1.2.7

Derived image usage:

FROM ghcr.io/mic1on/onestep-worker:1.2.7

WORKDIR /workspace
COPY . /workspace
ENV ONESTEP_TARGET=/workspace/worker.yaml

ONESTEP_TARGET points to the YAML file path or Python import target the container should start. The runtime automatically adds /workspace and /workspace/src to PYTHONPATH. If /workspace/requirements.txt exists it is installed first; otherwise the runtime falls back to installing /workspace when /workspace/pyproject.toml exists.

See deploy/worker-runtime-image.md for the full usage guide and troubleshooting notes.

Control Plane Reporter

onestep can push runtime telemetry to onestep-control-plane over a single long-lived WebSocket session without adding a new connector or changing task code.

Install the optional control-plane dependency before using the reporter:

pip install 'onestep[control-plane]'

The YAML form is intentionally minimal:

app:
  name: billing-sync

reporter: true

That enables ControlPlaneReporter using env-backed defaults. When you need explicit overrides, keep the field names aligned with ControlPlaneReporterConfig:

reporter:
  base_url: https://control-plane.example.com
  token: ${ONESTEP_CONTROL_PLANE_TOKEN}
  service_name: billing-sync-worker

Attach the reporter explicitly:

from onestep import ControlPlaneReporter, ControlPlaneReporterConfig, IntervalSource, OneStepApp

app = OneStepApp("billing-sync")
reporter = ControlPlaneReporter(ControlPlaneReporterConfig.from_env(app_name=app.name))
reporter.attach(app)


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

Required environment variables:

ONESTEP_CONTROL_PLANE_URL or ONESTEP_CONTROL_URL
ONESTEP_CONTROL_PLANE_TOKEN or ONESTEP_CONTROL_TOKEN

Common optional environment variables:

ONESTEP_ENV
ONESTEP_SERVICE_NAME
ONESTEP_NODE_NAME
ONESTEP_DEPLOYMENT_VERSION
ONESTEP_INSTANCE_ID
ONESTEP_REPLICA_KEY
ONESTEP_STATE_DIR
ONESTEP_CONTROL_PLANE_HEARTBEAT_INTERVAL_S
ONESTEP_CONTROL_PLANE_METRICS_INTERVAL_S
ONESTEP_CONTROL_PLANE_EVENT_FLUSH_INTERVAL_S
ONESTEP_CONTROL_PLANE_EVENT_BATCH_SIZE
ONESTEP_CONTROL_PLANE_MAX_PENDING_EVENTS
ONESTEP_CONTROL_PLANE_MAX_PENDING_METRIC_BATCHES
ONESTEP_CONTROL_PLANE_TIMEOUT_S
ONESTEP_CONTROL_PLANE_RECONNECT_BASE_DELAY_S
ONESTEP_CONTROL_PLANE_RECONNECT_MAX_DELAY_S

The reporter now uses:

GET /api/v1/agents/ws
hello / hello_ack
telemetry(sync|heartbeat|metrics|events)
command / command_ack / command_result

Behavior:

startup reuses a stable instance_id from local state unless ONESTEP_INSTANCE_ID or ONESTEP_REPLICA_KEY overrides it
startup opens a fresh WS session and negotiates capabilities
startup sends a heartbeat and a topology sync built from the current app tasks
sync and heartbeat sequences are tracked independently and persist across restarts
sync is resent on later heartbeat cycles until the current topology hash converges
task execution events are aggregated into task window metrics
important runtime events (retried, failed, dead_lettered, cancelled) are batched and pushed
remote commands can trigger ping, shutdown, restart, drain, pause_task, resume_task, sync_now, flush_metrics, and flush_events
transport send failures reset the current session and reconnect with exponential backoff plus jitter
low-priority metrics and events buffers are bounded locally; if the control plane stays down, the oldest buffered telemetry is dropped first
reporter failures are logged but do not stop task execution

Identity resolution order:

ONESTEP_INSTANCE_ID: hard override for tests or explicit pinning.
ONESTEP_REPLICA_KEY: derives a deterministic UUIDv5 from service_name + environment + replica_key.
local identity state: reuses the instance_id stored in the reporter state dir.

By default ControlPlaneReporterConfig.from_env() uses a local state dir under:

~/.onestep/control-plane-state/<environment>/<service_name>
~/.onestep/control-plane-state/<environment>/<service_name>/<replica_key> when ONESTEP_REPLICA_KEY is set

Multi-replica rules:

single worker: the default state dir is enough
multiple workers on one host: give each worker a unique ONESTEP_REPLICA_KEY or ONESTEP_STATE_DIR
StatefulSet-style deployments: use the stable ordinal as ONESTEP_REPLICA_KEY
disposable deployment pod names are not a stable replica key unless you inject one yourself

For an operations-focused guide with deployment patterns and troubleshooting, see docs/stable-instance-identity.md.

Quick local demo:

Start onestep-control-plane:

If the control plane repo is checked out next to this one:

cd ../onestep-control-plane
./scripts/start-local.sh

Start a long-running OneStep reporter demo:

./scripts/run-control-plane-demo.sh

Or run the end-to-end smoke script, which boots the local control plane, starts the demo agent, dispatches a ping, and waits for a successful command_result:

./scripts/run-control-plane-smoke.sh

Inspect the control plane. The demo cycles through ok, retry_once, fail, and slow jobs so you can see successful runs, retries, terminal failures, timeouts, and dead-letter events without changing any code:

http://127.0.0.1:8080/services?environment=dev
http://127.0.0.1:8080/services/control-plane-demo?environment=dev
http://127.0.0.1:8080/services/control-plane-demo?environment=dev&tab=commands

Runtime

The runtime now has a few explicit control points:

OneStepApp(..., shutdown_timeout_s=30.0) controls how long the app waits for inflight tasks before cancelling them during shutdown
@app.task(..., timeout_s=30.0) applies an execution timeout to async handlers
@app.task(..., dead_letter=...) routes terminal failures into a dead-letter sink
@app.on_event receives task execution events
InMemoryMetrics() is a built-in metrics hook for event counting
@app.on_startup and @app.on_shutdown register lifecycle hooks
ctx.config exposes app config
ctx.state exposes per-task namespaced state backed by the app state store

Failures are classified as:

error
timeout
cancelled

Custom retry policies receive a FailureInfo object so they can decide differently for timeouts vs business exceptions.

Execution events are emitted for:

fetched
started
succeeded
retried
failed
dead_lettered
cancelled

from onestep import InMemoryMetrics, InMemoryStateStore, MemoryQueue, OneStepApp, StructuredEventLogger

source = MemoryQueue("incoming")
dead = MemoryQueue("dead-letter")
metrics = InMemoryMetrics()
app = OneStepApp(
    "runtime-demo",
    config={"prefix": "demo"},
    state=InMemoryStateStore(),
    shutdown_timeout_s=10.0,
)
app.on_event(metrics)
app.on_event(StructuredEventLogger())


@app.on_startup
async def bootstrap(app):
    await source.publish({"value": 1})


@app.task(source=source, timeout_s=5.0, dead_letter=dead)
async def consume(ctx, item):
    runs = await ctx.state.get("runs", 0)
    await ctx.state.set("runs", runs + 1)
    print(ctx.config["prefix"], item)
    ctx.app.request_shutdown()

If a task ends in a terminal failure and dead_letter is configured, the dead-letter sink receives:

body["payload"]: the original payload
body["failure"]: {kind, exception_type, message, traceback?}
meta["original_meta"]: the original envelope metadata

You can inspect metrics in-process:

snapshot = metrics.snapshot()
print(snapshot["kinds"])

StructuredEventLogger() bridges TaskEvent into standard Python logging with consistent fields such as:

event_kind
app_name
task_name
source_name
attempts
duration_s
failure_kind

Tasks can also opt into webhook-friendly success metadata by returning a reserved notification object. The task return value sent to sinks stays unchanged; only the emitted succeeded event gains meta["notification"].

from onestep import MemoryQueue, OneStepApp, StructuredEventLogger

source = MemoryQueue("incoming")
app = OneStepApp("notification-demo")
app.on_event(StructuredEventLogger())


@app.task(source=source)
async def sync_status(ctx, item):
    updated = item["updated"]
    ctx.app.request_shutdown()
    return {
        "success": True,
        "updated": updated,
        "notification": {
            "summary": f"Status sync complete, updated {updated} devices",
            "metrics": [
                {"label": "Updated", "value": updated},
            ],
        },
    }

The emitted TaskEventKind.SUCCEEDED metadata then includes:

{
  "notification": {
    "summary": "Status sync complete, updated 12 devices",
    "metrics": [
      {"label": "Updated", "value": 12}
    ]
  }
}

notification values must be JSON-safe. Invalid values are dropped from the emitted metadata instead of failing task execution.

Memory Queue Example

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):
    ctx.app.request_shutdown()
    return {"value": item["value"] * 2}


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

Interval Source

Use a local scheduler when you need to run a task every fixed duration.

from onestep import IntervalSource, OneStepApp

app = OneStepApp("interval-demo")


@app.task(
    source=IntervalSource.every(
        hours=1,
        immediate=True,
        overlap="skip",
        payload={"job": "refresh-cache"},
    )
)
async def refresh_cache(ctx, item):
    print("scheduled at:", ctx.current.meta["scheduled_at"], "payload:", item)

overlap controls what happens when the previous run is still inflight:

allow: start another run immediately
skip: drop missed ticks while the previous run is still running
queue: serialize missed ticks and run them one by one afterwards

queue mode keeps at most max_queued_runs missed ticks, defaulting to 1000, and drops the oldest queued ticks first when the backlog exceeds that bound.

Cron Source

Use CronSource when you care about wall-clock time rather than elapsed duration.

from onestep import CronSource, OneStepApp

app = OneStepApp("hourly-sync")


@app.task(source=CronSource("0 * * * *", timezone="Asia/Shanghai", overlap="skip"))
async def sync_hourly(ctx, _):
    print("running at:", ctx.current.meta["scheduled_at"])

The built-in parser supports standard 5-field cron expressions and these aliases:

@hourly
@daily
@weekly
@monthly
@yearly

Webhook Source

Use WebhookSource when external systems push events into your app.

from onestep import BearerAuth, MemoryQueue, OneStepApp, WebhookSource

app = OneStepApp("webhook-demo")
jobs = MemoryQueue("jobs")


@app.task(
    source=WebhookSource(
        path="/webhooks/github",
        methods=("POST",),
        host="127.0.0.1",
        port=8080,
        auth=BearerAuth("replace-me"),
    ),
    emit=jobs,
)
async def ingest_github(ctx, event):
    return {
        "event": event["headers"].get("x-github-event"),
        "payload": event["body"],
    }

The stdlib implementation supports:

shared host:port listeners across multiple webhook routes
optional BearerAuth(...)
json, form, text, raw, and auto body parsing
fixed WebhookResponse(...) responses
exact path matching and method filtering

The payload delivered to your task contains:

body
headers
query
method
path
client
received_at

MySQL Table Queue

Use a table as a task queue by claiming rows and marking them as finished.

from onestep import MemoryQueue, OneStepApp
from onestep_mysql import MySQLConnector

app = OneStepApp("orders")
db = MySQLConnector("mysql+pymysql://root:root@localhost:3306/app")
source = db.table_queue(
    table="orders",
    key="id",
    where="status = 0",
    claim={"status": 9},
    ack={"status": 1},
    nack={"status": 0},
    batch_size=100,
)
sink = db.table_sink(table="processed_orders", mode="upsert", keys=("id",))


@app.task(source=source, emit=sink, concurrency=16)
async def process_order(ctx, row):
    return {"id": row["id"], "payload": row["payload"], "status": "done"}

When you need to write computed fields back to the claimed row itself, call await ctx.update_current_row({...}) inside the task. This is currently supported for MySQLConnector.table_queue(...) deliveries.

MySQL Incremental Sync

Use (updated_at, id) as a lightweight cursor for Logstash-style sync.

from onestep import MemoryQueue, OneStepApp
from onestep_mysql import MySQLConnector

app = OneStepApp("sync-users")
db = MySQLConnector("mysql+pymysql://root:root@localhost:3306/app")
state = db.cursor_store(table="onestep_cursor")
source = db.incremental(
    table="users",
    key="id",
    cursor=("updated_at", "id"),
    where="deleted = 0",
    batch_size=1000,
    state=state,
)
out = MemoryQueue("dw")


@app.task(source=source, emit=out, concurrency=1)
async def sync_user(ctx, row):
    return {"id": row["id"], "name": row["name"], "updated_at": row["updated_at"]}

For production deployments, prefer db.cursor_store(...) or db.state_store(...) over the in-memory stores so cursors and task state survive process restarts.

Feishu Bitable Plugin

Install onestep-feishu-bitable to use Feishu Bitable as an incremental source or upsert sink. In this repository, the plugin source lives under plugins/onestep-feishu-bitable.

from onestep import OneStepApp
from onestep_feishu_bitable import (
    FeishuBitableConnector,
    feishu_bitable_text,
    feishu_bitable_user,
)

app = OneStepApp("feishu-sync")
feishu = FeishuBitableConnector(app_id="cli_xxx", app_secret="secret")
source = feishu.incremental(
    app_token="bascnxxx",
    table_id="tbl_source",
    cursor_field="最后更新时间",
    user_id_type="user_id",
    batch_size=100,
    fallback_scan_page_limit=100,
)
sink = feishu.table_sink(
    app_token="bascnyyy",
    table_id="tbl_target",
    mode="upsert",
    match_fields=["编号"],
    user_id_type="user_id",
)


@app.task(source=source, emit=sink, concurrency=20)
async def sync_order(ctx, payload):
    fields = payload["fields"]
    return {
        "编号": feishu_bitable_text(fields.get("编号")),
        "标题": feishu_bitable_text(fields.get("标题")),
        "负责人": feishu_bitable_user(fields.get("负责人ID")),
    }

For Feishu person fields, pass the matching user_id_type (open_id, union_id, or user_id) and write values as [{"id": "..."}]. The feishu_bitable_text(...) and feishu_bitable_user(...) helpers live in the onestep_feishu_bitable plugin package.

When Feishu cannot serve the incremental search with cursor sorting, the source falls back to scanning pages and sorting locally. fallback_scan_page_limit bounds that fallback to avoid high memory use on large tables.

RabbitMQ Queue

from onestep import OneStepApp
from onestep_rabbitmq import RabbitMQConnector

app = OneStepApp("rabbitmq-demo")
rmq = RabbitMQConnector("amqp://guest:guest@localhost/")
source = rmq.queue(
    "incoming_jobs",
    exchange="jobs.events",
    routing_key="jobs.created",
    prefetch=50,
)
out = rmq.queue(
    "processed_jobs",
    exchange="jobs.events",
    routing_key="jobs.done",
)


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

Install with pip install onestep-mq.

Redis Streams

Use Redis Streams for lightweight, reliable message queuing with consumer groups.

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,
    poll_interval_s=0.5,
)
out = redis.stream("processed")


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

Key features:

Consumer groups: Multiple consumers share message processing
Message acknowledgment: XACK for reliable processing
Pending messages: Unacked messages stay in PEL for retry via XCLAIM
Stream trimming: maxlen option to limit stream size

Install with pip install onestep-redis.

SQS Queue

from onestep import OneStepApp
from onestep_sqs import SQSConnector

app = OneStepApp("sqs-demo")
sqs = SQSConnector(region_name="ap-southeast-1")
source = sqs.queue(
    "https://sqs.ap-southeast-1.amazonaws.com/123456789/jobs.fifo",
    message_group_id="workers",
    delete_batch_size=10,
    delete_flush_interval_s=0.5,
    heartbeat_interval_s=15,
    heartbeat_visibility_timeout=60,
)
out = sqs.queue(
    "https://sqs.ap-southeast-1.amazonaws.com/123456789/processed.fifo",
    message_group_id="workers",
)


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

Install with pip install onestep-sqs.

Examples

Supported examples are indexed in:

example/README.md

Common entrypoints:

example/cli_app.py
example/runtime_showcase.py
example/mysql_incremental.py
example/redis_stream.py
example/webhook_source.py

Integration Tests

Optional live tests are under tests/integration/ and plugin-specific plugins/*/tests/integration/ directories. The local stack now includes Redis, RabbitMQ, LocalStack SQS, and MySQL.

Install the live-test dependencies:

pip install '.[integration]'

Start the local integration stack:

make integration-up

Load the generated environment into your current shell:

eval "$(./scripts/setup-integration-env.sh)"

Run all live tests in one command:

make integration-test

You can also run one test file manually after loading the environment:

PYTHONPATH=src python3 -m pytest tests/integration/test_redis_live.py -q
PYTHONPATH=src python3 -m pytest tests/integration/test_rabbitmq_live.py -q
uv run --all-packages python -m pytest plugins/onestep-sqs/tests/integration/test_sqs_live.py -q
uv run --all-packages python -m pytest plugins/onestep-mysql/tests/integration/test_mysql_live.py -q

Set KEEP_INTEGRATION_SERVICES=1 to keep containers running after make integration-test.

Test Layout

The test suite is now intentionally split by responsibility:

tests/contract/: runtime contract tests that lock task execution semantics
tests/integration/: live infrastructure tests for built-in Redis and RabbitMQ connectors
plugins/*/tests/: plugin-specific unit, contract, and live integration tests
tests/test_*.py: connector-focused unit tests

End-to-End Demo

For a quick end-to-end demo with webhook ingestion, queueing, dead-letter handling, metrics, and structured logs:

PYTHONPATH=src python3 example/runtime_showcase.py

Then send:

curl -X POST http://127.0.0.1:8090/demo/webhook \
  -H 'Content-Type: application/json' \
  -d '{"action":"ok","value":21}'

curl -X POST http://127.0.0.1:8090/demo/webhook \
  -H 'Content-Type: application/json' \
  -d '{"action":"fail","value":21}'

curl -X POST http://127.0.0.1:8090/demo/webhook \
  -H 'Content-Type: application/json' \
  -d '{"action":"slow","value":21}'

You will see:

structured task event logs
successful processing for action=ok
dead-letter output for action=fail
timeout + dead-letter output for action=slow

Name		Name	Last commit message	Last commit date
Latest commit History 343 Commits
.github/workflows		.github/workflows
.vscode		.vscode
deploy		deploy
docker/worker		docker/worker
docs		docs
example		example
plugins		plugins
scripts		scripts
skills/onestep		skills/onestep
src/onestep		src/onestep
tests		tests
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
LICENSE		LICENSE
MIGRATION-0.5-to-1.0.0.md		MIGRATION-0.5-to-1.0.0.md
Makefile		Makefile
README.md		README.md
docker-compose.integration.yml		docker-compose.integration.yml
pyproject.toml		pyproject.toml
uv.lock		uv.lock

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Repository files navigation

onestep

Install

Upgrading from 0.5.x

CLI

systemd

Official worker image

Control Plane Reporter

Runtime

Memory Queue Example

Interval Source

Cron Source

Webhook Source

MySQL Table Queue

MySQL Incremental Sync

Feishu Bitable Plugin

RabbitMQ Queue

Redis Streams

SQS Queue

Examples

Integration Tests

Test Layout

End-to-End Demo

About

Uh oh!

Releases 68

Uh oh!

Contributors

Uh oh!

Languages

Folders and files

Latest commit

History

Repository files navigation

onestep

Install

Upgrading from 0.5.x

CLI

systemd

Official worker image

Control Plane Reporter

Runtime

Memory Queue Example

Interval Source

Cron Source

Webhook Source

MySQL Table Queue

MySQL Incremental Sync

Feishu Bitable Plugin

RabbitMQ Queue

Redis Streams

SQS Queue

Examples

Integration Tests

Test Layout

End-to-End Demo

About

Topics

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases 68

Uh oh!

Contributors

Uh oh!

Languages