orionis.foundation.application

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

module documentation

(source)

Undocumented

Async Function	`_asgi_receive_dispatcher`	Consume ASGI receive messages and dispatch them concurrently.
Async Function	`_await_http_tasks`	Await the kernel and finish its disconnect monitor before returning.
Function	`_create_http_handler_task`	Start a monitored handler through the event loop's task factory.
Function	`_parse_config_key`	Parse a dot-notated configuration key into immutable path components.
Async Function	`_rsgi_disconnect_watcher`	Cancel the kernel task when the RSGI disconnect awaitable completes.
Constant	`_ASGI_BODY_QUEUE_SIZE`	Undocumented
Constant	`_CONFIG_KEY_CACHE_SIZE`	Undocumented
Constant	`_CWD`	Undocumented
Constant	`_ERR_NOT_CONFIGURED`	Undocumented
Constant	`_MAINTENANCE_REFRESH_NS`	Undocumented
Constant	`_ROUTER_FACADE_MODULE`	Undocumented
Constant	`_SENTINEL`	Undocumented

async def _asgi_receive_dispatcher(receive: Callable[[], Awaitable[dict[str, Any]]], request_queue: asyncio.Queue, handler_task: asyncio.Task) -> bool: (source) ¶

Consume ASGI receive messages and dispatch them concurrently.

Forward body chunks through a bounded queue and cancel the handler when the receive channel reports a disconnect or fails. A full queue pauses receive calls until the handler consumes a buffered message.

Parameters
receive:`Callable[[], Awaitable[dict[str, Any]]]`	ASGI receive callable provided by the server.
request_queue:`asyncio.Queue`	Queue that buffers body messages for the handler.
handler_task:`asyncio.Task`	Kernel task to cancel when the receive channel closes.

Returns
`bool`	Whether the disconnect requested cancellation of the handler.

async def _await_http_tasks(handler_task: asyncio.Task, disconnect_task: asyncio.Task[bool]) -> Any: (source) ¶

Await the kernel and finish its disconnect monitor before returning.

Parameters
handler_task:`asyncio.Task`	Kernel task handling the request.
disconnect_task:`asyncio.Task[bool]`	Monitor returning whether it cancelled the kernel on disconnect.

Returns
`Any`	Kernel result, or None after a client disconnect.

Raises
`asyncio.CancelledError`	If the caller or kernel is cancelled independently of a disconnect.
`Exception`	If the kernel or disconnect monitor fails.

def _create_http_handler_task(loop: asyncio.AbstractEventLoop, coroutine: Coroutine) -> asyncio.Task: (source) ¶

Start a monitored handler through the event loop's task factory.

Parameters
loop:`asyncio.AbstractEventLoop`	Running request event loop.
coroutine:`Coroutine`	Kernel handler coroutine to schedule.

Returns
`asyncio.Task`	Handler task. The native loop without a custom factory starts it eagerly; other loops and factories receive their standard task-creation arguments.

@lru_cache(maxsize=_CONFIG_KEY_CACHE_SIZE)

def _parse_config_key(key: str) -> tuple[str, ...]: (source) ¶

Parse a dot-notated configuration key into immutable path components.

Parameters
key:`str`	Validated configuration key whose empty segments must be preserved.

Returns
`tuple[str, ...]`	Path components shared independently of application configuration values.

async def _rsgi_disconnect_watcher(protocol: HTTPProtocol, handler_task: asyncio.Task) -> bool: (source) ¶

Cancel the kernel task when the RSGI disconnect awaitable completes.

Parameters
protocol:`HTTPProtocol`	Protocol providing the client disconnect awaitable.
handler_task:`asyncio.Task`	Kernel task associated with the connection.

Returns
`bool`	Whether the disconnect requested cancellation of the handler.

_ASGI_BODY_QUEUE_SIZE: int = (source) ¶

Undocumented

Value

_CONFIG_KEY_CACHE_SIZE: int = (source) ¶

Undocumented

Value

_CWD = (source) ¶

Undocumented

Value

Path.cwd()

_ERR_NOT_CONFIGURED: str = (source) ¶

Undocumented

Value

'Application configuration is not initialized. Please call create() first.'

_MAINTENANCE_REFRESH_NS: int = (source) ¶

Undocumented

Value

100000000

_ROUTER_FACADE_MODULE: str = (source) ¶

Undocumented

Value

'orionis.support.facades.router'

_SENTINEL = (source) ¶

Undocumented

Value

object()