DOCSEARCHbyAlgolia
Submit Docs

Wove: Inline Task Orchestration for Python

repository·main·Indexed 42 days ago

https://github.com/curvedinf/wove

A Python library for running high-latency async tasks concurrently with an improved developer experience over standard asyncio. Wove uses the weave() context manager to build a dependency graph based on function signatures, automatically managing task parallelism and dependencies. Version 2.0.3 supports local inline concurrency, background processing in embedded or forked modes, and remote task environments.

Tokens
59.7K
Snippets
174
Records
255
Agent score
52%

What's inside Wove

  1. What is an Environment in Wove

    main

    An environment is a persistent, named execution profile. While individual weaves are lightweight and non-persistent, environments store durable configuration regarding where work runs, concurrency limits, and delivery policies.

    Use environments to:

    • Offload specific tasks to external systems (queues, clusters, batch schedulers) while keeping others local.
    • Apply shared defaults like retries, timeout, or max_workers across a project.
    • Decouple backend infrastructure configuration from your task code.
    • Switch infrastructure globally by updating wove.config(...) instead of modifying individual decorators.

  2. How task mapping works in Wove

    main

    Task mapping allows you to run the same task once for each item in an iterable. Wove executes these task instances concurrently, collects all results into a list, and passes that list to downstream tasks using the mapped task's result name.

    Use task mapping when a workflow step needs to 'fan out' over multiple inputs, such as:

    • Fetching a list of URLs
    • Transforming a set of records
    • Parsing multiple files
    • Loading specific IDs

  3. Use Backend Adapters to run tasks in existing infrastructure

    main

    Use a backend adapter when you want Wove tasks to run inside infrastructure your project already uses, such as a queue, workflow engine, cluster, or batch system (e.g., Celery, Temporal, Ray, Kubernetes, AWS Batch, Slurm).

    Unlike network executors that talk directly to Wove workers via HTTP/gRPC, backend adapters allow your existing task system to own the delivery path, worker placement, and scheduling. The task code itself does not need to import the backend library; you simply configure the environment in Wove.

    import wove
from myapp.reports import build_report, load_account
from wove import weave

wove.config(
    default_environment="default",
    environments={
        "default": {"executor": "local"},
        "reports": {
            "executor": "celery",
            "executor_config": {
                "broker_url": "redis://redis:6379/0",
                "task_name": "myapp.wove_task",
            },
            "delivery_timeout": 30.0,
        },
    },
)

with weave() as w:
    @w.do
    def account():
        return load_account()

    @w.do(environment="reports")
    def report(account):
        return build_report(account)

  4. Inspect failed results without raising exceptions

    main

    By default, accessing a failed result (via w.result.task_name, w.result["task_name"], or w.result.final) will re-raise the task's exception.

    To treat failures as data instead of exceptions (useful for dashboards or reports), set error_mode="return" in the weave() call. In this mode, accessing a failed task result returns the exception object itself. Note that result.exception always contains the first exception Wove recorded, regardless of the error_mode setting.

    from wove import weave

with weave(error_mode="return") as w:
    @w.do
    def fails():
        raise ValueError("bad input")

value = w.result.fails

assert isinstance(value, ValueError)
assert str(value) == "bad input"
assert w.result.exception is value

  5. How the RQ execution flow works

    main

    The RQ adapter follows a specific execution lifecycle:

    1. Enqueue: Wove enqueues a task using wove.integrations.worker.run(payload) into the Redis queue.
    2. Execution: An RQ worker picks up the task and executes the run function.
    3. Callback: After task completion, the worker sends a POST request with the results back to the configured callback_url.

  6. Implement custom executors or use built-in executors

    main

    Wove provides a public executor interface for building custom execution engines.

    Built-in executor types include:

    • Local execution: Runs tasks in the current process.
    • stdio subprocess: Stream-based execution via standard input/output.
    • Network executor: Delivers tasks over the network via:
      • HTTP/HTTPS
      • gRPC
      • WebSocket

    Network executors support security features including signed requests, bearer tokens, and TLS. For development, an explicit insecure configuration is available.

  7. Understand the `grpc` Executor protocol and implementation

    main

    The grpc executor uses gRPC's generic unary call support, meaning Wove does not require generated client stubs.

    Request/Response Format

    • Request: Serialized Wove command bytes.
    • Response: Serialized JSON containing either a single event frame, a list of event frames, or an object with an events list.

    Worker Service Requirements

    The remote service must implement the configured method such that it accepts raw command bytes and returns raw event bytes. A run_task response should eventually include one of the following terminal event frames:

    • task_result
    • task_error
    • task_cancelled

    Long-running services can emit task_started before returning the terminal event when the unary call completes.

    Note: If your worker requires continuous heartbeats or long-lived streaming, use the websocket executor instead.

  8. Use `wove.helpers` for data-shaping tasks

    main
    The wove.helpers module provides small data-shaping utilities designed to make task dependencies easier to express within a Wove weave. Use these helpers when you need to perform simple transformations—such as flattening iterables, grouping data, or handling optional values—that belong inside the weave logic but do not require a dedicated, complex task function.

  9. How the HTTP Executor protocol works

    main

    The HTTP executor acts as a transport layer that sends Wove command frames to a remote worker service via HTTP POST requests.

    Request Behavior

    For every run_task or cancel_task command, Wove POSTs a JSON body to the configured url. Python objects that cannot be directly encoded in JSON are moved into dispatch fields such as callable_pickle and args_pickle.

    Security Requirements

    • Production/Remote: Use https:// and set security="env:WOVE_WORKER_SECRET" to ensure signed requests and TLS.
    • Local Development: Plain http:// without security is permitted only for local endpoints, or if executor_config.insecure=True is explicitly set.

    Worker Service Response Format

    The worker service must respond with JSON. The response can be a single event frame, a list of event frames, or an object containing an events list. A run_task response must eventually include one of the following terminal frames: task_result, task_error, or task_cancelled.

    {
  "events": [
    {"type": "task_started", "run_id": "...", "task_id": "render_report"},
    {"type": "task_result", "run_id": "...", "task_id": "render_report", "result_pickle": "..."}
  ]
}

  10. Configure Wove Runtime and Environments

    main

    Wove manages execution through a process-wide configuration singleton and environment resolution.

    • Runtime Configuration: Use wove.runtime to manage process-wide settings and wove.config(...) to apply configurations.
    • Environments: Persistent execution profiles are defined via environment dictionaries. Wove handles precedence, defaults, and validation rules to resolve these profiles during runtime.
    • Security: For network-based execution, wove.security provides helpers for request signing and verification across HTTP, gRPC, and WebSocket executors.

  11. How the Wove backend callback flow works

    main

    Wove uses a dispatch pattern to offload tasks from a local weave to external task systems (like Celery, Temporal, Ray, Kubernetes, AWS Batch, or Slurm). The flow follows these steps:

    1. Initialization: BackendAdapterEnvironmentExecutor starts a BackendCallbackServer in the local process.
    2. Payload Creation: Wove builds a dispatch payload from a run_task frame containing the serialized callable, arguments, task/run identities, delivery settings, adapter name, and a callback_url.
    3. Submission: The adapter submits this payload to the backend system.
    4. Worker Execution: A backend worker receives the payload and invokes wove.integrations.worker.run(payload) or await wove.integrations.worker.arun(payload).
    5. Event Posting: The worker executes the task and posts status events (task_started, task_result, task_error, or task_cancelled) to the callback_url.
    6. Resolution: BackendCallbackServer.recv() captures these frames and returns them to the executor runtime, allowing the original weave to resolve the pending task.

    Note: The callback_url points to the Wove callback server. You can control where Wove listens using callback_host and callback_port, and control the URL embedded in payloads using callback_url.

  12. Implement a custom `stdio` worker process

    main

    A stdio executor works by communicating via Wove's JSON-lines frame protocol over standard streams. If you are building a custom worker process, it must adhere to these responsibilities:

    1. Read from stdin: Read one JSON command frame per line from stdin.
    2. Decode: Decode task callables and arguments from the serialized frame fields.
    3. Execute: Execute or forward the task.
    4. Write to stdout: Write one JSON event frame per line to stdout.
    5. Error Handling: Emit task_error frames for any task or transport failures.