Reference

Engin

The Engin is a modular application defined by a collection of options.

Users should instantiate the Engin with a number of options, where options can be an instance of Provide, Invoke, or a collection of these combined in a Block.

To create a useful application, users should pass in one or more providers (Provide or Supply) and at least one invocation (Invoke or Entrypoint).

When instantiated the Engin can be run. This is typically done via the run method, but for advanced usecases it can be easier to use the start and stop methods.

When ran the Engin takes care of the complete application lifecycle:

1. The Engin assembles all Invocations. Only Providers that are required to satisfy the Invoke options parameters are assembled.
2. All Invocations are run sequentially in the order they were passed in to the Engin.
3. Lifecycle Startup tasks registered by assembled dependencies are run sequentially.
4. The Engin waits for a stop signal, i.e. SIGINT or SIGTERM, or a supervised task that causes a shutdown.
5. Lifecyce Shutdown tasks are run in the reverse order to the Startup order.

Examples:

import asyncio

from httpx import AsyncClient

from engin import Engin, Invoke, Lifecycle, Provide


def httpx_client(lifecycle: Lifecycle) -> AsyncClient:
    client = AsyncClient()
    lifecycle.append(client)
    return client


async def main(http_client: AsyncClient) -> None:
    print(await http_client.get("https://httpbin.org/get"))

engin = Engin(Provide(httpx_client), Invoke(main))

asyncio.run(engin.run())

Examples:

>>> engin = Engin(Provide(construct_a), Invoke(do_b), Supply(C()), MyBlock())

Parameters:

Name	Type	Description	Default
*options	Option	an instance of Provide, Supply, Invoke, Entrypoint or a Block.	()

run async

run() -> None

Run the engin.

The engin will run until it is stopped via an external signal (i.e. SIGTERM or SIGINT), the stop method is called on the engin, or a lifecycle task errors.

start async

start() -> None

Starts the engin in the background. This method will wait until the engin is fully started to return so it is safe to use immediately after.

stop async

stop() -> None

Stop the engin.

This method will wait for the shutdown lifecycle to complete before returning. Note this method can be safely called at any point, even before the engin is started.

graph

graph() -> list[Node]

Creates a graph representation of the engin's dependencies which can be used for introspection or visualisations.

Returns: a list of Node objects.

Provide

Bases: Dependency[Any, T]

Parameters:

Name	Type	Description	Default
factory	Func[P, T]	the factory function that returns the type.	required
scope	str | None	(optional) associate this provider with a specific scope.	None
as_type	type | None	(optional) allows you to explicitly specify the provided type, e.g. to type erase a concrete type, or to provide a mock implementation.	None
override	bool	(optional) allow this provider to override other providers for the same type from the same package.	False

Lifecycle

Allows dependencies to define startup and shutdown tasks for the application.

Lifecycle tasks are defined using Context Managers, these can be async or sync.

Lifecycle tasks should generally be defined in Providers as they are tied to the construction of a given dependency, but can be used in Invocations. The Lifecycle type is provided as a built-in Dependency by the Engin framework.

Examples:

Using a type that implements context management.

from httpx import AsyncClient

def my_provider(lifecycle: Lifecycle) -> AsyncClient:
    client = AsyncClient()

    # AsyncClient is a context manager
    lifecycle.append(client)

Defining a custom lifecycle.

def my_provider(lifecycle: Lifecycle) -> str:
    @contextmanager
    def task():
        print("starting up!")
        yield
        print("shutting down!)

    lifecycle.append(task)

Defining a custom lifecycle using a LifecycleHook.

def my_provider(lifecycle: Lifecycle) -> str:
    connection_pool = ConnectionPool()

    lifecycle.hook(
        on_start=connection_pool.connect,
        on_stop=connection_pool.close,
    )

append

append(cm: _AnyContextManager) -> None

Append a Lifecycle task to the list.

Parameters:

Name	Type	Description	Default
cm	_AnyContextManager	a task defined as a ContextManager or AsyncContextManager.	required

hook

hook(
    *,
    on_start: _ParameterlessCallable | None = None,
    on_stop: _ParameterlessCallable | None = None,
) -> None

Append a hook to the Lifecycle.

At least one of on_start or on_stop must be provided.

Parameters:

Name	Type	Description	Default
on_start	_ParameterlessCallable | None	a callable to be executed on Lifecycle startup.	None
on_stop	_ParameterlessCallable | None	a callable to be executed on Lifecycle shutdown.	None

list

list() -> list[AbstractAsyncContextManager]

List all the defined tasks.

Returns:

Type	Description
list[AbstractAsyncContextManager]	A copy of the list of Lifecycle tasks.