Reference

Public API

The following are the main exports from qbox:

qbox.observe(obj, scope=None)

Observe an object, collapsing it if it’s a QBox.

If obj is a QBox: forces evaluation, replaces references, returns value. If obj is not a QBox: returns obj unchanged (idempotent).

Observing a composed QBox observes its entire dependency tree.

Parameters:

  • obj (T | QBox[T]) – The object to observe (may or may not be a QBox).

  • scope (Literal['locals', 'stack', 'globals'] | None) – Override the default scope for replacement. If None, uses the QBox’s default scope.

Returns:

The unwrapped value if obj is a QBox, otherwise obj unchanged.

Return type:

T

Example:

from qbox import QBox, observe
data = QBox(fetch_data())
value = observe(data)  # Forces evaluation
value = observe("hello")  # Returns "hello" unchanged
class qbox.QBox(awaitable, mimic_type=None, scope='stack', start='soon', repr_observes=False, cancel_on_delete=True)

A lazily awaited container for async operations.

QBox wraps a coroutine or awaitable and defers its evaluation until the value is actually needed. Operations on a QBox return new QBox instances, allowing lazy composition of async operations.

The value is computed on a background thread running an asyncio event loop. When the value is needed (e.g., for comparison, str conversion, or explicit access via __wrapped__), the QBox blocks until the result is available.

Parameters:

  • awaitable (Awaitable[T]) – A coroutine or awaitable to wrap.

  • mimic_type (type[T] | None)

  • scope (ScopeType)

  • start (StartMode)

  • repr_observes (bool)

  • cancel_on_delete (bool)

Return type:

QBox[T]

Example:

async def fetch_value() -> int:
    return 42

box = QBox(fetch_value())
result = box + 8  # Returns QBox, no blocking
if result > 40:   # Blocks here, evaluates to 50
    print("Large!")
__await__()

Support awaiting the QBox in async contexts.

Yields:

The result of the async operation.

Return type:

Generator[Any, None, T]

static __new__(cls, awaitable, mimic_type=None, scope='stack', start='soon', repr_observes=False, cancel_on_delete=True)

Create a new QBox instance.

If mimic_type is provided, returns an instance of a typed QBox subclass that is registered with the appropriate ABC for isinstance support.

Parameters:

  • awaitable (Awaitable[T]) – The coroutine or awaitable to wrap (used by __init__).

  • mimic_type (type[T] | None) – Optional type hint for ABC registration.

  • scope (ScopeType) – Scope for reference replacement (used by __init__).

  • start (StartMode) – When to start execution (‘soon’ or ‘observed’).

  • repr_observes (bool) – Whether repr() triggers observation.

  • cancel_on_delete (bool) – Whether to cancel pending work on deletion.

Returns:

A QBox instance (possibly of a typed subclass).

Return type:

QBox[T]

property __wrapped__: T

Get the wrapped value, blocking if necessary.

This property triggers evaluation of the async operation if it hasn’t been evaluated yet. If the operation raised an exception, that exception is re-raised here.

Returns:

The result of the async operation.

Raises:

Exception – Any exception raised by the wrapped coroutine.

qbox.ScopeType

alias of Literal[‘locals’, ‘stack’, ‘globals’]

qbox.StartMode

alias of Literal[‘soon’, ‘observed’]

isinstance Support

Functions for transparent isinstance behavior with QBox:

qbox.enable_qbox_isinstance()

Patch builtins.isinstance to work transparently with QBox.

After calling this, isinstance(qbox, SomeType) will: 1. Force observation of the QBox 2. Replace references to the QBox with the unwrapped value 3. Return the correct isinstance result

This is a global change affecting all code. Call once at startup.

Note

  • This function is idempotent (safe to call multiple times)

  • Use disable_qbox_isinstance() to restore original behavior

  • Consider using ABCs with mimic_type instead for library code

  • This function is thread-safe

Return type:

None

qbox.disable_qbox_isinstance()

Restore the original isinstance behavior.

After calling this, isinstance will return False for QBox instances when checking against the wrapped type (normal Python behavior).

Note

  • This function is idempotent (safe to call multiple times)

  • This function is thread-safe

Return type:

None

qbox.is_qbox_isinstance_enabled()

Check if isinstance has been patched for QBox transparency.

Returns:

True if enable_qbox_isinstance() has been called and disable_qbox_isinstance() has not been called since.

Return type:

bool

qbox.qbox_isinstance()

Context manager for scoped QBox isinstance behavior.

Within this context, isinstance() will work transparently with QBox instances, forcing observation and returning the correct result.

This context manager supports nesting - the patching will remain active until the outermost context exits.

Example

>>> from qbox import QBox, qbox_isinstance
>>> async def fetch_dict():
...     return {"key": "value"}
>>> data = QBox(fetch_dict())
>>> with qbox_isinstance():
...     assert isinstance(data, dict)  # True, forces observation
>>> # Original isinstance restored after block

>>> # Nested contexts work correctly:
>>> with qbox_isinstance():
...     with qbox_isinstance():
...         pass  # Still enabled
...     # Still enabled after inner context exits
>>> # Disabled after outermost context exits

Note

  • This function is thread-safe

  • Nested contexts increment a reference count; patching is only disabled when the outermost context exits

Return type:

Generator[None, None, None]

Internal Modules

These modules are implementation details but documented for completeness.

qbox.qbox

The Quantum Box container - a lazily awaited async container.

class qbox.qbox.QBox(awaitable, mimic_type=None, scope='stack', start='soon', repr_observes=False, cancel_on_delete=True)

Bases: Generic[T]

A lazily awaited container for async operations.

QBox wraps a coroutine or awaitable and defers its evaluation until the value is actually needed. Operations on a QBox return new QBox instances, allowing lazy composition of async operations.

The value is computed on a background thread running an asyncio event loop. When the value is needed (e.g., for comparison, str conversion, or explicit access via __wrapped__), the QBox blocks until the result is available.

Parameters:

  • awaitable (Awaitable[T]) – A coroutine or awaitable to wrap.

  • mimic_type (type[T] | None)

  • scope (ScopeType)

  • start (StartMode)

  • repr_observes (bool)

  • cancel_on_delete (bool)

Return type:

QBox[T]

Example:

async def fetch_value() -> int:
    return 42

box = QBox(fetch_value())
result = box + 8  # Returns QBox, no blocking
if result > 40:   # Blocks here, evaluates to 50
    print("Large!")

qbox._loop

Background event loop manager for running async operations.

class qbox._loop.BackgroundLoopManager

Singleton managing a background asyncio event loop.

This class maintains a single daemon thread running an asyncio event loop that can execute coroutines from synchronous code. The loop runs until the Python interpreter shuts down.

The singleton pattern ensures all QBox instances share the same background loop, avoiding thread proliferation.

Return type:

BackgroundLoopManager

property loop: AbstractEventLoop

Get the background event loop.

submit(coro)

Submit a coroutine for execution on the background loop.

Parameters:

coro (Coroutine[Any, Any, T]) – The coroutine to execute.

Returns:

A concurrent.futures.Future that will contain the result.

Return type:

Future[T]

qbox._loop.get_loop_manager()

Get the global BackgroundLoopManager instance.

Returns:

The singleton BackgroundLoopManager.

Return type:

BackgroundLoopManager

qbox._loop.submit_to_loop(coro)

Submit a coroutine for execution on the background event loop.

The background loop runs on a separate daemon thread, which ensures that blocking on the result (via future.result()) never deadlocks, regardless of whether the caller is in a sync or async context.

Parameters:

coro (Coroutine[Any, Any, T]) – The coroutine to execute.

Returns:

A concurrent.futures.Future that will contain the result.

Return type:

Future[T]

qbox._isinstance

Opt-in transparent isinstance support for QBox.

This module provides functions to patch builtins.isinstance to work transparently with QBox instances. When enabled, isinstance(box, SomeType) will force observation of the QBox and return the correct result.

Example usage:

from qbox import enable_qbox_isinstance, disable_qbox_isinstance

# Enable at startup
enable_qbox_isinstance()

# Now isinstance works transparently
data = QBox(fetch_dict())
isinstance(data, dict)  # True! Forces observation automatically

# Optionally disable later
disable_qbox_isinstance()

Or use the context manager for scoped patching:

with qbox_isinstance():
    isinstance(data, dict)  # True in this block
# Original isinstance restored after block
qbox._isinstance.disable_qbox_isinstance()

Restore the original isinstance behavior.

After calling this, isinstance will return False for QBox instances when checking against the wrapped type (normal Python behavior).

Note

  • This function is idempotent (safe to call multiple times)

  • This function is thread-safe

Return type:

None

qbox._isinstance.enable_qbox_isinstance()

Patch builtins.isinstance to work transparently with QBox.

After calling this, isinstance(qbox, SomeType) will: 1. Force observation of the QBox 2. Replace references to the QBox with the unwrapped value 3. Return the correct isinstance result

This is a global change affecting all code. Call once at startup.

Note

  • This function is idempotent (safe to call multiple times)

  • Use disable_qbox_isinstance() to restore original behavior

  • Consider using ABCs with mimic_type instead for library code

  • This function is thread-safe

Return type:

None

qbox._isinstance.is_qbox_isinstance_enabled()

Check if isinstance has been patched for QBox transparency.

Returns:

True if enable_qbox_isinstance() has been called and disable_qbox_isinstance() has not been called since.

Return type:

bool

qbox._isinstance.qbox_isinstance()

Context manager for scoped QBox isinstance behavior.

Within this context, isinstance() will work transparently with QBox instances, forcing observation and returning the correct result.

This context manager supports nesting - the patching will remain active until the outermost context exits.

Example

>>> from qbox import QBox, qbox_isinstance
>>> async def fetch_dict():
...     return {"key": "value"}
>>> data = QBox(fetch_dict())
>>> with qbox_isinstance():
...     assert isinstance(data, dict)  # True, forces observation
>>> # Original isinstance restored after block

>>> # Nested contexts work correctly:
>>> with qbox_isinstance():
...     with qbox_isinstance():
...         pass  # Still enabled
...     # Still enabled after inner context exits
>>> # Disabled after outermost context exits

Note

  • This function is thread-safe

  • Nested contexts increment a reference count; patching is only disabled when the outermost context exits

Return type:

Generator[None, None, None]