orionis.aio.Loop

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class Loop: (source)

View In Hierarchy

Thread-safe, platform-aware asyncio event loop manager for the Orionis Framework.

Centralises every aspect of event loop lifecycle:

Optimal factory selection: uvloop (non-Windows) → ProactorEventLoop (Windows) → stdlib default.
Per-thread loop creation and caching to prevent cross-thread sharing.
Transparent bridging between synchronous and asynchronous execution contexts without deadlocking a running loop.
Cooperative task cancellation and cleanup on context-manager exit.

All state is class-level; no instance creation is required or intended.

Class Method	`_detectUvloop`	Detect and cache the uvloop event loop factory, if available.
Class Method	`_getLoopFactory`	Return the best available event loop factory for the current platform.
Class Method	`_getSyncExecutor`	Return the shared single-worker thread pool used for sync↔async bridging.
Class Method	`getEventLoop`	Return the event loop for the current thread, creating one if necessary.
Class Method	`runSync`	Run a coroutine synchronously from any context.
Static Method	`_getRunningLoop`	Return the event loop currently running in this thread, or `None`.
Async Static Method	`createTask`	Create and schedule a new asyncio task for coro.
Static Method	`eventLoopContext`	Context manager that provides an event loop and cleans up on exit.
Async Static Method	`execute`	Transparently execute a sync or async callable.
Static Method	`isLoopRunning`	Return `True` if an event loop is currently running in the calling thread.
Static Method	`run`	Execute a coroutine as the application entry point.
Constant	`_IS_WIN32`	Undocumented
Class Variable	`_loop_factory_cached`	Undocumented
Class Variable	`_loop_factory_resolved`	Undocumented
Class Variable	`_loop_local`	Undocumented
Class Variable	`_loop_lock`	Undocumented
Class Variable	`_sync_executor`	Undocumented
Class Variable	`_sync_executor_lock`	Undocumented
Class Variable	`_uvloop_checked`	Undocumented
Class Variable	`_uvloop_factory`	Undocumented

@classmethod

def _detectUvloop(cls) -> Callable[[], asyncio.AbstractEventLoop] | None: (source) ¶

Detect and cache the uvloop event loop factory, if available.

Uses double-checked locking so detection occurs at most once across all threads; subsequent calls return the cached result immediately.

Returns
`Callable or None`	`uvloop.new_event_loop` when available on a non-Windows platform, otherwise `None`.

@classmethod

def _getLoopFactory(cls) -> Callable[[], asyncio.AbstractEventLoop] | None: (source) ¶

Return the best available event loop factory for the current platform.

Resolution order:

uvloop (non-Windows only).
asyncio.ProactorEventLoop (Windows only).
None — the caller falls back to asyncio.new_event_loop().

The result is cached after the first call; repeated invocations are essentially free.

Returns
`Callable or None`	Undocumented

@classmethod

def _getSyncExecutor(cls) -> concurrent.futures.ThreadPoolExecutor: (source) ¶

Return the shared single-worker thread pool used for sync↔async bridging.

The pool is created lazily on first access via double-checked locking and reused for all subsequent calls, keeping thread-creation overhead off hot paths.

Returns
`concurrent.futures.ThreadPoolExecutor`	Undocumented

@classmethod

def getEventLoop(cls) -> asyncio.AbstractEventLoop: (source) ¶

Return the event loop for the current thread, creating one if necessary.

If a loop is already running in the calling thread it is returned immediately. Otherwise the per-thread cached loop is returned; a fresh loop is created and registered when no valid cached one exists.

Returns
`asyncio.AbstractEventLoop`	Undocumented

@classmethod

def runSync(cls, coro: Coroutine[Any, Any, T]) -> T: (source) ¶

Run a coroutine synchronously from any context.

When no loop is running, delegates directly to run. When a loop is already running (e.g. inside an async framework), the coroutine is dispatched to the shared single-worker thread pool so it runs its own loop without deadlocking the caller.

Parameters
coro:`Coroutine`

Returns
`Any`	Undocumented

@staticmethod

def _getRunningLoop() -> asyncio.AbstractEventLoop | None: (source) ¶

Return the event loop currently running in this thread, or None.

Uses the public asyncio.get_running_loop() API with a try/except to avoid relying on CPython private internals. The overhead is negligible when a loop is present (fast path).

Returns
`asyncio.AbstractEventLoop or None`	Undocumented

@staticmethod

async def createTask(coro: Coroutine[Any, Any, T], *, name: str | None = None) -> asyncio.Task[T]: (source) ¶

Create and schedule a new asyncio task for coro.

Parameters
coro:`Coroutine`	The coroutine to schedule.
name:`str or None, optional`	An optional descriptive name for the task.

Returns
`asyncio.Task`	Undocumented

@staticmethod
@contextmanager

def eventLoopContext() -> Generator[asyncio.AbstractEventLoop]: (source) ¶

Context manager that provides an event loop and cleans up on exit.

Pending tasks are cancelled cooperatively and awaited with return_exceptions=True so no exception escapes the finally block.

Yields
`asyncio.AbstractEventLoop`	Undocumented

@staticmethod

async def execute(func: Callable[..., Any], /, *args: Any, **kwargs: Any) -> Any: (source) ¶

Transparently execute a sync or async callable.

Async callables are awaited directly; synchronous callables are offloaded to the event loop's default executor to avoid blocking the loop thread.

Parameters
func:`Callable`	The function or coroutine function to invoke.
*args:`Any`	Positional arguments forwarded to func.
**kwargs:`Any`	Keyword arguments forwarded to func.

Returns
`Any`	Undocumented

Raises
`TypeError`	If func is not callable.

@staticmethod

def isLoopRunning() -> bool: (source) ¶

Return True if an event loop is currently running in the calling thread.

Returns
`bool`	Undocumented

@staticmethod

def run(coro: Coroutine[Any, Any, T]) -> T: (source) ¶

Execute a coroutine as the application entry point.

Designed to be called from a context with no running event loop (e.g. CLI __main__). Passes KeyboardInterrupt cleanly so the process exits with code 0 on Ctrl+C.

Parameters
coro:`Coroutine`	The coroutine object to run.

Returns
`Any`	The value returned by the coroutine, or `0` on `KeyboardInterrupt`.

Raises
`TypeError`	If coro is not a coroutine object.
`RuntimeError`	Raised when a loop is already running in the calling thread. The message matches the relevant standard library entry point (`asyncio.Runner` or `asyncio.run`); coro is left unconsumed. Use `runSync` to bridge into a running loop.

_IS_WIN32: ClassVar[bool] = (source) ¶

Undocumented

Value

sys.platform == 'win32'

_loop_factory_cached: ClassVar[Callable[[], asyncio.AbstractEventLoop] | None] = (source) ¶

Undocumented

_loop_factory_resolved: ClassVar[bool] = (source) ¶

Undocumented

_loop_local: ClassVar[threading.local] = (source) ¶

Undocumented

_loop_lock: ClassVar[threading.Lock] = (source) ¶

Undocumented

_sync_executor: ClassVar[concurrent.futures.ThreadPoolExecutor | None] = (source) ¶

Undocumented

_sync_executor_lock: ClassVar[threading.Lock] = (source) ¶

Undocumented

_uvloop_checked: ClassVar[bool] = (source) ¶

Undocumented

_uvloop_factory: ClassVar[Callable[[], asyncio.AbstractEventLoop] | None] = (source) ¶

Undocumented