API reference

Applications and shared pool

Independent threaded asyncio applications sharing one Python runtime.

UniteIO gives each application class one singleton instance, one daemon thread, and one asyncio event loop. All applications share the singleton UIOPool for synchronous callable work while coroutine work remains on the application’s own loop. On free-threaded CPython, callbacks belonging to different loop threads can execute Python code in parallel when their workload permits it. Process-wide objects remain shared and require normal thread-safe access when mutable.

Example

Define, start, submit work to, and stop an application:

import asyncio
from uniteio import UniteIO

class Service(UniteIO, prefix="SVC"):
    async def __call__(self):
        await asyncio.Event().wait()

service = Service()
future = service.submit(asyncio.sleep, 0, result="ready")
assert future.result() == "ready"
service.stop()

The package targets CPython free-threaded builds. Separate loops still provide scheduling isolation with the GIL enabled, but the GIL prevents parallel Python-bytecode execution between their threads. Closing an application loop uses a CPython-specific executor-detachment step so the shared pool remains available to other applications.

class uniteio.uniteio.UniteIO(**kwargs)

Bases: UIOArchetype

Base class for singleton applications running dedicated asyncio loops.

Each concrete subclass owns one active daemon thread and event loop. On a free-threaded CPython build, work on different application threads can run in parallel while sharing process-wide state; mutable shared state must be synchronized by the application. The subclass’s asynchronous __call__() method is scheduled automatically during construction. Synchronous callables submitted through submit() run on the shared UIOPool; coroutines run on the application’s loop.

Keyword arguments accepted by a subclass can be forwarded to this base constructor. Keys that do not already exist become both attributes and item-accessible configuration values.

Parameters:: **kwargs – Application configuration exposed through attributes and app[key] access.

Example

Create an application, submit both kinds of work, and stop it:

import asyncio
from uniteio import UniteIO

class Processor(UniteIO, prefix="PROC"):
    def __init__(self, endpoint):
        super().__init__(endpoint=endpoint)

    async def __call__(self):
        await asyncio.Event().wait()

app = Processor(endpoint="local")
assert app.endpoint == "local"
assert app["endpoint"] == "local"
assert app.submit(sum, [1, 2, 3]).result() == 6
app.stop()

taskname()

Return a unique task name using the application’s prefix.

Returns:: A name such as "API:3".
Return type:: str

__init__(**kwargs)

Initialize thread state, lifecycle events, and configuration.

Parameters:: **kwargs – Configuration values to expose as attributes and through __getitem__(). Existing attributes are not overwritten.
Return type:: None

run()

Run and ultimately close the application’s event loop.

This is the target executed by threading.Thread; callers should not invoke it directly. On exit it cancels pending tasks, finalizes asynchronous generators, detaches the shared executor, closes the loop, and unregisters the application.

Return type:: None

stop(wait=True, timeout=None)

Stop this application while leaving the shared executor active.

When called outside the application thread, wait=True waits for task cancellation, loop closure, and thread termination. Calls made by the application itself never block waiting for their own thread.

Parameters:

wait (bool) – Whether an external caller should join the application thread before returning. Ignored when called by the application itself.
timeout (float | None) – Maximum seconds to wait for thread termination, or None to wait indefinitely.

Returns:

True when teardown has completed before return; otherwise False. A self-stop request returns False because cleanup continues after the current callback yields control.

Return type:

bool

Example

Request shutdown and wait at most two seconds:

if not app.stop(timeout=2):
    print("application is still stopping")

async asleep(delay, result=None)

Sleep asynchronously on the application’s event loop.

Parameters:

delay – Number of seconds to suspend the current coroutine.
result – Value returned after the delay. A falsey value selects the application instance instead.

Returns:

result when truthy, otherwise self.

Example

await self.asleep(0.1, "ready") returns "ready".

sleep(delay, result=None)

Block the current thread for a specified duration.

Use this helper only in synchronous work submitted to UIOPool; calling it on an event-loop thread blocks that loop.

Parameters:

delay – Number of seconds to block the current thread.
result – Value returned after the delay. A falsey value selects the application instance instead.

Returns:

result when truthy, otherwise self.

Example

app.submit(app.sleep, 0.1, "ready").result() returns "ready".

exception_handler(loop, context)

Record an asyncio exception context and delegate default reporting.

Parameters:

loop (BaseEventLoop) – Event loop that reported the exception.
context – Asyncio exception-handler context dictionary.

Return type:

None

Example

Install the handler from run() with loop.set_exception_handler(self.exception_handler).

submit(target: Coroutine, /, *, name: str | None = None, eager: bool = False) → Future

submit(target: Iterable[Coroutine], /, *, name: str | None = None, eager: bool = False) → Future

Submit coroutine or synchronous work to the appropriate runtime.

Coroutine objects and coroutine functions execute on this application’s event loop. An iterable of coroutine objects executes inside an asyncio.TaskGroup. Other callables execute on the shared UIOPool.

Parameters:

target (Callable) – Coroutine object, coroutine function, iterable of coroutine objects, or synchronous callable.
*args – Positional arguments passed to a coroutine function or synchronous callable.
name (str | None) – Optional asyncio task name. Batch child names append their zero-based index, for example "batch:0".
context (Context) – Optional contextvars.Context used for coroutine execution. It does not apply to synchronous executor work.
eager (bool) – Request eager task start when supported by asyncio.
**kwargs – Keyword arguments passed to a coroutine function or synchronous callable.

Returns:

A concurrent.futures.Future. A batch future resolves to an insertion-ordered mapping from each original coroutine object to its return value.

Raises:

TypeError – If target is unsupported or a batch contains a non-coroutine member.
RuntimeError – If coroutine work is submitted after the application event loop has closed, or synchronous work is submitted after the shared executor has shut down.

Return type:

Future

Example

Submit coroutine and synchronous work from another thread:

async def fetch(identifier):
    return identifier

one = app.submit(fetch, 1, name="fetch-one").result()
many = app.submit(
    [fetch(2), fetch(3)],
    name="fetch-batch",
).result()
total = app.submit(sum, [one, *many.values()]).result()

class uniteio.uniteio.UIOPool(max_workers=None)

Bases: AdjustableThreadPoolExecutor

Process-wide executor shared by active UniteIO applications.

The pool is a restartable singleton: repeated construction returns the active executor, while construction after shutdown() creates a new executor. Each application still owns a distinct asyncio event loop.

Parameters:: max_workers – Maximum number of executor worker threads. None uses the standard ThreadPoolExecutor default.

Example

Submit synchronous work directly to the shared pool:

from uniteio import UIOPool

pool = UIOPool(max_workers=4)
assert pool.submit(sum, [1, 2, 3]).result() == 6
pool.shutdown()

__init__(max_workers=None)

Initialize the shared executor and application registry.

Parameters:: max_workers – Maximum number of worker threads, or None to use the standard executor default.

discard(app)

Remove an application without affecting the executor.

Parameters:: app (UIOArchetype) – Application to remove. A stale or unknown instance is ignored.
Return type:: None

shutdown(wait=True, *, cancel_futures=False)

Stop every application before shutting down the executor.

Stop requests are broadcast to all registered applications first, then every application thread is joined. Only after their loops are closed is ThreadPoolExecutor shutdown invoked. The pool is removed from the singleton registry after successful completion.

Parameters:

wait – Passed to ThreadPoolExecutor.shutdown. Applications are always joined first, even when this argument is False.
cancel_futures – Whether queued executor futures that have not begun execution should be cancelled.

Raises:

RuntimeError – If called from one of the registered application threads, because that thread cannot synchronously join itself.

Example

Stop all applications and replace the exhausted singleton:

old_pool = UIOPool()
old_pool.shutdown(cancel_futures=True)
new_pool = UIOPool()
assert new_pool is not old_pool

Adjustable executor

Thread-pool executor whose worker ceiling can be increased at runtime.

The implementation intentionally relies on CPython’s private ThreadPoolExecutor state. It changes the ceiling used when deciding whether new workers may be created; it never terminates existing workers.

class uniteio.adjustable_pool.AdjustableThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=(), **ctxkwargs)

Bases: ThreadPoolExecutor

A ThreadPoolExecutor with a mutable worker-creation ceiling.

Lowering the ceiling is allowed only when the requested value is not below the number of worker threads that already exist. Existing workers are never retired.

Parameters:: max_workers – Initial maximum number of worker threads. All constructor arguments are inherited from ThreadPoolExecutor.

Example

Increase a running executor from two workers to four:

from uniteio import AdjustableThreadPoolExecutor

with AdjustableThreadPoolExecutor(max_workers=2) as pool:
    old, new = pool.set_max_workers(4)
    assert (old, new) == (2, 4)

property max_workers: int

Return the current worker-creation ceiling.

Returns:: The maximum number of threads the executor may create.

set_max_workers(value)

Change the ceiling used for future worker creation.

This method does not create workers immediately and does not retire workers that already exist.

Parameters:

value (int) – New positive integer worker ceiling. Objects implementing __index__ are accepted; booleans are rejected explicitly.

Returns:

A (previous, current) tuple containing the old and new limits.

Raises:

TypeError – If value is not an integer or is a boolean.
ValueError – If value is not positive or is smaller than the number of workers already created.
RuntimeError – If the executor has shut down or is broken.

Return type:

tuple[int, int]

Example

Resize an executor before submitting additional work:

pool = AdjustableThreadPoolExecutor(max_workers=1)
try:
    pool.set_max_workers(3)
    result = pool.submit(pow, 2, 8).result()
finally:
    pool.shutdown()