autoasync 0.1.3

Retrofit concurrency into synchronous Python code: start work in the background now, block only when the result is actually used.

autoasync

Wrap synchronous work so it starts in the background and returns immediately.
The call only blocks when you actually use the result.

from autoasync import autoasync


def load_report(path):
    ...


load_report_async = autoasync(load_report)
report = load_report_async("report.csv")

prepare_page()                    # runs while load_report works in background
print(report)                     # blocks here only if the result is not ready yet

Why use this instead of async / await?

autoasync is mainly for retrofitting concurrency into existing synchronous code. Compared with Python's built-in async / await, its main advantages are:

• No async contagion: callers can stay synchronous, so you do not need to turn the whole call chain into async def just to overlap one slow step.
• Minimal code changes: wrapping a function with @autoasync or autoasync(fn) is often enough to start work in the background.
• Deferred waiting: you do not have to decide upfront where to await; execution only blocks when the value is actually needed.
• Easy incremental optimization: this is convenient when improving a mature sync codebase, because you can add concurrency without redesigning APIs around an event loop.

That trade-off is intentional: this abstraction adds overhead, so it is not a replacement for native async / await in high-throughput async systems. It is most useful when you want a simple, low-friction way to hide latency in otherwise synchronous code.

Install

pip install autoasync

Usage

Decorate any synchronous function

from autoasync import autoasync


@autoasync
def fetch(url: str) -> str:
    import requests
    return requests.get(url).text


result = fetch("https://example.com")
do_other_work()
print(result)

Run several calls concurrently

a = fetch("https://example.com/a")
b = fetch("https://example.com/b")
c = fetch("https://example.com/c")

combined = a + "\n" + b + "\n" + c

CPU-bound work with processes

use_process=True is only supported for importable module-level functions.

from autoasync import autoasync


@autoasync(use_process=True)
def crunch(n: int) -> int:
    return sum(range(n))


result = crunch(10_000_000)
do_other_work()
print(result)

Custom executor

from concurrent.futures import ThreadPoolExecutor
from autoasync import autoasync

pool = ThreadPoolExecutor(max_workers=4)


@autoasync(executor=pool)
def read(path: str) -> str:
    with open(path) as f:
        return f.read()

Configure built-in executors globally

Use configure_autoasync(...) when you want to keep the library-managed executors but control their default sizes.

from autoasync import autoasync, configure_autoasync

configure_autoasync(thread_max_workers=8, process_max_workers=4)

@autoasync
def fetch(url):
    ...


@autoasync(use_process=True)
def crunch(n):
    return sum(range(n))


page = fetch("https://example.com")   # uses the configured thread pool
total = crunch(10_000_000)            # uses the configured process pool

Configuration changes apply to future built-in pools only. Existing cached pools are not replaced. If you pass a custom executor, the global built-in pool configuration is ignored.

Use reset_autoasync() to clear cached built-in pools and restore default configuration, which is especially useful in tests.

Access the original function

@autoasync
def add(a, b):
    return a + b


add.__wrapped__(1, 2)

What the returned value supports

The returned value behaves like the final result for most common Python protocols, including:

• attribute access
• arithmetic and comparisons
• container access and iteration
• conversions such as int, float, str, bytes, and bool
• context managers
• open() path usage through __fspath__

The background result is resolved once and then cached.

Important behavior

is is special

is checks object identity and cannot be overloaded in Python. That means:

resolve_true = autoasync(lambda: True)
result = resolve_true()

result == True         # works
bool(result) is True   # works
result is True         # False: `is` checks identity, not the eventual value

If you need identity-style checks, resolve first or use == / bool(...).

Exceptions are deferred

If the wrapped function raises an exception, the exception is not raised at call time.
It is re-raised when the result is first needed.

@autoasync
def explode():
    raise RuntimeError("boom")


result = explode()     # no exception yet
print(result)          # raises RuntimeError here

Process mode restrictions

When use_process=True, the wrapped function must be:

• defined at module scope
• importable from its module
• a normal function, not a lambda, nested function, bound method, or callable object
• optionally paired with configure_autoasync(process_max_workers=...) to size the built-in process pool

Unsupported targets raise a clear TypeError.

How it works

result = slow_fn(arg)          # submits fn(arg) to an executor
                               # returns immediately

do_other_work()                # runs concurrently with fn(arg)

print(result + 1)              # blocks only here if needed

Requirements

• Python 3.8+
• No runtime dependencies

License

MIT