Component

elva.component

Module for generic asynchronous app component.

Classes:

• Component –

  Generic asynchronous app component class.

Component

Generic asynchronous app component class.

This class features graceful shutdown alongside annotated logging. It is used for writing providers, stores, parsers, renderers etc.

It supports explicit handling via the start and stop method as well as the asynchronous context manager protocol.

Methods:

• start –

  Start the component.

• stop –

  Stop the component by cancelling all inner task groups.

• before –

  Hook to run before the component signals that is has been started.

• run –

  Hook to run after the component signals that is has been started.

• cleanup –

  Hook to run after the component's stop method

Attributes:

• log (Logger) –

  Logger instance to write logging messages to.

• started (Event) –

  Event signaling that the component has been started, i.e. is initialized and running.

• stopped (Event) –

  Event signaling that the component has been stopped, i.e. deinitialized and not running.

log instance-attribute

Logger instance to write logging messages to.

started property

Event signaling that the component has been started, i.e. is initialized and running.

stopped property

Event signaling that the component has been stopped, i.e. deinitialized and not running.

start(task_status=TASK_STATUS_IGNORED) async

Start the component.

Parameters:

• task_status (TaskStatus[None], default: TASK_STATUS_IGNORED ) –

  The status to set when the task has started.

Source code in src/elva/component.py

async def start(self, task_status: TaskStatus[None] = TASK_STATUS_IGNORED):
    """
    Start the component.

    Arguments:
        task_status: The status to set when the task has started.
    """
    self.log.info("starting")
    async with self._get_start_lock():
        if self._task_group is not None:
            raise RuntimeError(f"{self} already running")

        async with create_task_group() as self._task_group:
            await self._task_group.start(self._run)
            task_status.started()

stop() async

Stop the component by cancelling all inner task groups.

Source code in src/elva/component.py

async def stop(self):
    """
    Stop the component by cancelling all inner task groups.
    """
    if self._task_group is None:
        raise RuntimeError(f"{self} not running")

    self._task_group.cancel_scope.cancel()
    self.log.debug("cancelled")

before() async

Hook to run before the component signals that is has been started.

In here, one would define initializing steps necessary for the component to run. This method must return, otherwise the component will not set the started signal.

It is defined as a no-op and supposed to be implemented in the inheriting class.

Source code in src/elva/component.py

async def before(self):
    """
    Hook to run before the component signals that is has been started.

    In here, one would define initializing steps necessary for the component to run.
    This method must return, otherwise the component will not set the
    [`started`][elva.component.Component.started] signal.

    It is defined as a no-op and supposed to be implemented in the inheriting class.
    """
    ...

run() async

Hook to run after the component signals that is has been started.

In here, one would define the main functionality of the component. This method may run indefinitely or return. The component is kept running regardless.

It is defined as a no-op and supposed to be implemented in the inheriting class.

Source code in src/elva/component.py

async def run(self):
    """
    Hook to run after the component signals that is has been started.

    In here, one would define the main functionality of the component.
    This method may run indefinitely or return.
    The component is kept running regardless.

    It is defined as a no-op and supposed to be implemented in the inheriting class.
    """
    ...

cleanup() async

Hook to run after the component's stop method has been called and before it sets the stopped event.

In here, one would define cleanup tasks such as closing connections. This method must return, otherwise the component will not set the stopped signal.

It is defined as a no-op and supposed to be implemented in the inheriting class.

Source code in src/elva/component.py

async def cleanup(self):
    """
    Hook to run after the component's [`stop`][elva.component.Component.stop] method
    has been called and before it sets the [`stopped`][elva.component.Component.stopped] event.

    In here, one would define cleanup tasks such as closing connections.
    This method must return, otherwise the component will not set the
    [`stopped`][elva.component.Component.stopped] signal.

    It is defined as a no-op and supposed to be implemented in the inheriting class.
    """
    ...