cmeta.core_async module

CMeta asynchronous wrapper for the core class

See the cMeta COPYRIGHT and LICENSE files in the project root for details.

Classes

class cmeta.core_async.CMetaAsync(max_workers=None, logger=None, loop=None, **kwargs)[source]

Bases: CMeta

Async wrapper around CMeta using multiprocessing for non-blocking operations.

This class extends CMeta to provide asynchronous wrappers around blocking operations using a ProcessPoolExecutor. It prevents blocking the event loop in FastAPI or other asyncio-based services by executing CMeta operations in separate worker processes.

Inherits all CMeta methods while providing async wrappers for blocking operations.

__init__(max_workers=None, logger=None, loop=None, **kwargs)[source]

Initialize CMetaAsync with process pool executor.

Parameters:

max_workers – Maximum number of worker processes in the pool.
logger – Custom logger instance. If None, uses parent CMeta’s logger.
loop – Event loop to use. If None, gets the current event loop.
**kwargs – Additional keyword arguments passed to CMeta constructor.

Returns:

Operation result.

Return type:

dict

Raises:

Exception – Propagated runtime errors, if any.

async access(params)[source]

Asynchronous non-blocking wrapper for CMeta.access().

Runs CMeta.access() in a worker process to avoid blocking the event loop.

Parameters:: params – Dictionary of parameters to pass to CMeta.access().
Returns:: Result dictionary from CMeta.access() with ‘return’ and other keys.
Return type:: dict
Raises:: Exception – Propagated runtime errors, if any.

access_sync(params)[source]

Synchronous access: the command runs in this process, blocking, on a plain CMeta of this process - made at the first call with the constructor arguments of this object, as a worker process makes its own - so a command that calls cm.access() itself (most of them) gets a result dict, as it does in a worker.

Use this method when already executing in a worker thread or process where blocking is acceptable, or before an event loop runs (at import). It never runs the command on this async object: until KI-001 was fixed it did, and a nested cm.access() landed on the async access() and handed back an un-awaited coroutine in place of the result.

Parameters:: params – Dictionary of parameters to pass to CMeta.access().
Returns:: Result dictionary from CMeta.access().
Return type:: dict
Raises:: Exception – Propagated runtime errors, if any.

shutdown()[source]

Gracefully shutdown the process pool executor.

Waits for all pending tasks to complete before shutting down the executor. This method can be tied to FastAPI or other framework shutdown events.

Parameters:: None.
Returns:: Operation result.
Return type:: dict
Raises:: Exception – Propagated runtime errors, if any.