What is Wove For?

Wove is for running high latency async tasks like web requests and database queries concurrently in the same way as asyncio, but with a drastically improved user experience. Improvements compared to asyncio include:

Reads Top-to-Bottom: The code in a weave block is declared in the order it is executed inline in your code instead of in disjointed functions.
Implicit Parallelism: Parallelism and execution order are implicit based on function and parameter naming.
Sync or Async: Mix async def and def freely. A weave block can be inside or outside an async context. Sync functions are run in a background thread pool to avoid blocking the event loop.
Normal Python Data: Wove's task data looks like normal Python variables because it is. This is because of inherent multithreaded data safety produced in the same way as map-reduce.
Automatic Scheduling: Wove builds a dependency graph from your task signatures and runs independent tasks concurrently as soon as possible.
Automatic Detachment: Wove can run your inline code in a forked detached process so you can return your current process back to your server's pool.
Extensibility: Define parallelized workflow templates that can be overridden inline.
High Visibility: Wove includes debugging tools that allow you to identify where exceptions and deadlocks occur across parallel tasks, and inspect inputs and outputs at each stage of execution.
Minimal Boilerplate: Get started with just the with weave() as w: context manager and the @w.do decorator.
Fast: Wove has low overhead and internally uses asyncio, so performance is comparable to using threading or asyncio directly.
Free Threading Compatible: Running a modern GIL-less Python? Build true multithreading easily with a weave.

Installation

Download wove with pip:

pip install wove

The Basics

The core of Wove's functionality is the weave context manager. It is used in a with block to define a list of tasks that will be executed as concurrently and as soon as possible. When Python closes the weave block, the tasks are executed immediately based on a dependency graph that Wove builds from the function signatures. Results of a task are passed to any same-named function parameters. The result of the last task that runs are available in w.result.final.

import time
from wove import weave
with weave() as w:
    # These first two tasks run concurrently.
    @w.do
    def magic_number():
        time.sleep(1.0)
        return 42
    @w.do
    def important_text():
        time.sleep(1.0)
        return "The meaning of life"
    # This task depends on the first two. It runs only after both are complete.
    @w.do
    def combined(important_text, magic_number):
        return f"{important_text} is {magic_number}!"
    # When the `with` block closes, all tasks are executed.
print(w.result.final)
# >> The meaning of life is 42!
print(f"The magic number was {w.result.magic_number}")
# >> The magic number was 42
print(f'The important text was "{w.result["important_text"]}"')
# >> The important text was "The meaning of life"

Wove's Design Pattern

Wove is designed to be added inline in your existing functions. Since it is not required to be in an async block, it is useful for retrofitting into any IO-bound parallelizable process. For instance in a non-async Django view, you can run your database lookups and related code in parallel.

# views.py
import time
from django.shortcuts import render
from wove import weave
from .models import Author, Book

def author_details(request, author_id):
    with weave() as w:
        # `author` and `books` run concurrently
        @w.do
        def author():
            return Author.objects.get(id=author_id)
        @w.do
        def books():
            return list(Book.objects.filter(author_id=author_id))

        # Map the books to a task that updates each of their prices concurrently
        @w.do("books", retries=3)
        def books_with_prices(book):
            book.get_price_from_api()
            return book

        # When everything is done, create the template context
        @w.do
        def context(author, books_with_prices):
            return {
                "author": author,
                "books": books_with_prices,
            }
    return render(request, "author_details.html", w.result.final)

We suggest naming weave tasks with nouns instead of verbs. Since weave tasks are designed to be run immediately like inline code, and not be reused, noun names reinforce the concept that a weave task represents its output data instead of its action.

Core API

The two core Wove tools are:

weave(): An async context manager used in either a with or async with block that creates the execution environment for your tasks. When the weave block ends, all tasks will be executed in the order of their dependency graph. The weave object has a result attribute that contains the results of all tasks and a .final attribute that contains the result of the last task.
@w.do: A decorator that registers a function as a task to be run within the weave block. It can be used on both def and async def functions interchangeably, with non-async functions being run in a background thread pool to avoid blocking the event loop. It can optionally be passed an iterable, and if so, the task will be run concurrently for each item in the iterable. It can also be passed a string of another task's name, and if so, the task will be run concurrently for each item in the iterable result of the named task.

More Spice

This example demonstrates Wove's advanced features, including inheritable overridable Weaves, static task mapping, dynamic task mapping, merging an external function, and a complex task graph.

import time
import numpy as np
from wove import Weave, weave, merge

# An external function that will be mapped with `merge`
def quality_check(data):
    return any(np.isnan(data))

# Define a reusable base pipeline
class DataPipeline(Weave):
    @Weave.do(retries=2, timeout=60.0)
    def dataset(self, records: int):
        # Initial data loading - the top of the diamond.
        # `records` is provided by the `weave()` call below.
        time.sleep(0.1)
        return np.linspace(0, 10, records)
    
    @Weave.do("dataset")
    def feature_a(self, item):
        # First parallel processing branch.
        time.sleep(0.2)
        return np.sin(item)
    
    @Weave.do("dataset")
    def feature_b(self, item):
        # Second parallel processing branch.
        time.sleep(0.3)
        return np.cos(item)
    
    @Weave.do
    def merged_features(self, feature_a, feature_b):
        # Merge the results from parallel branches - bottom of the diamond.
        return np.column_stack((feature_a, feature_b))
    
    @Weave.do
    async def report(self, merged_features):
        # Dynamically map an external function using `merge`.
        quality_result = any(await merge(quality_check, merged_features))
        quality_status = "WARN: NaN values detected" if quality_result else "OK"
        # Create a report from the merged features.
        return {
            "mean": float(np.mean(merged_features)),
            "std": float(np.std(merged_features)),
            "shape": merged_features.shape,
            "quality_status": quality_status,
        }
# The class isn't executed right now

# Run a customized version of the pipeline
with wea

Wove

Install / Use

README

Table of Contents