Install & Compatibility
Where this runs
tested against v0.2.9 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.920 runs
installs and imports cleanly · install 0.0s · import 0.072s · 18MB
glibcpy 3.10–3.920 runs
installs and imports cleanly · install 1.6s · import 0.063s · 18MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
run
✓ from duet import run
The primary entry point to execute an async function with Duet.
AwaitableFuture
✓ from duet import AwaitableFuture
Used to wrap standard `concurrent.futures.Future` objects to make them awaitable within Duet's async context.
This quickstart demonstrates how to use Duet to run asynchronous code, wrap standard `concurrent.futures.Future` objects using `AwaitableFuture`, and showcases Duet's unique re-entrancy feature by calling `duet.run` within an already running async context.
import concurrent.futures
import time
import duet
def long_running_sync_task(value):
"""A blocking synchronous task."""
time.sleep(0.1) # Simulate work
return f"Processed {value}"
async def async_wrapper(future: concurrent.futures.Future):
"""Wraps a concurrent.futures.Future to be awaitable in Duet."""
return await duet.AwaitableFuture(future)
async def main():
print("Starting Duet example...")
# Example 1: Running a synchronous task in a thread pool executor
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(long_running_sync_task, "data_item_A")
result_a = await async_wrapper(future)
print(f"Result 1: {result_a}")
# Example 2: Demonstrate re-entrancy (calling duet.run within an async function)
# In most async libraries (e.g., asyncio, trio), this would raise a RuntimeError
# Duet explicitly allows it for easier refactoring of mixed sync/async code.
async def nested_task():
print(" Running nested_task synchronously via duet.run...")
nested_future = executor.submit(long_running_sync_task, "data_item_B")
nested_result = await async_wrapper(nested_future)
print(f" Nested task result: {nested_result}")
return "Nested task completed"
nested_call_result = duet.run(nested_task())
print(f"Result 2: {nested_call_result}")
print("Duet example finished.")
if __name__ == "__main__":
duet.run(main())
Errors
Common errors & fixes
TypeError: object Future can't be awaited
Attempting to `await` a `concurrent.futures.Future` object directly without wrapping it in `duet.AwaitableFuture`.
fixImport `AwaitableFuture` from `duet` and wrap the future: `await duet.AwaitableFuture(your_future)`.
NameError: name 'run' is not defined
The `run` function, or other Duet symbols, were used without being properly imported from the `duet` library.
fixEnsure you have imported the necessary components, e.g., `from duet import run` or `from duet import AwaitableFuture`.
Upgrade
Version history
0.2.9latest on PyPI · released Jul 26, 2023
Audit
Dependencies
pythonrequiredRequired Python version for the library.