rats.apps

rats.apps

Libraries to help create applications with a strong focus on composability and testability.

Applications give you the ability to define a development experience to match your project's domain.

AppPlugin = _AppPluginType | Callable[[Container], AppContainer] module-attribute

Main interface for a function that returns an rats.apps.AppContainer instance.

Functions that act as runners or application factories often take this as their input type in order to manage the top most rats.apps.Container instance, allowing most containers to use the rats.apps.PluginMixin mixin and rely on the top most container being managed automatically. This is the companion type to rats.apps.ContainerPlugin.

ContainerPlugin = _ContainerPluginType | Callable[[Container], Container] module-attribute

Main interface for a function that returns an rats.apps.Container instance.

Containers that implement this type—for example, by using the rats.apps.PluginMixin mixin—can be used easily in functions that need to defer the construction of an application container. See rats.apps.AppBundle for additional examples.

T_ExecutableType = TypeVar('T_ExecutableType', bound=Executable) module-attribute

T_ServiceType = TypeVar('T_ServiceType') module-attribute

__all__ = ['App', 'AppBundle', 'AppContainer', 'AppPlugin', 'CompositeContainer', 'CompositePlugin', 'Container', 'ContainerPlugin', 'DuplicateServiceError', 'Executable', 'GroupProvider', 'NullRuntime', 'PluginContainers', 'PluginMixin', 'Provider', 'ProviderNamespaces', 'Runtime', 'ServiceId', 'ServiceNotFoundError', 'StandardRuntime', 'StaticContainer', 'StaticProvider', 'T_ExecutableType', 'T_ServiceType', 'autoid', 'autoid_factory_service', 'autoid_service', 'autoscope', 'container', 'factory_service', 'fallback_group', 'fallback_service', 'group', 'run', 'run_plugin', 'service', 'static_group', 'static_service'] module-attribute

factory_service = _FactoryService module-attribute

App(callback)

Bases: Executable

Wraps a plain callable objects as a rats.apps.Executable.

This simple object allows for turning any callable object into an executable that is recognized by the rest of the rats application.

Example

from rats import apps

apps.App(lambda: print("hello, world")).execute()

Created by providing a reference to a Callable[[], None] function.

Parameters:

Name	Type	Description	Default
callback	Callable[[], None]	called when the application instance is executed.	required

Source code in rats/apps/_executables.py

def __init__(self, callback: Callable[[], None]) -> None:
    """
    Created by providing a reference to a `Callable[[], None]` function.

    Args:
        callback: called when the application instance is executed.
    """
    self._callback = callback

execute()

Runs the provided callback.

Source code in rats/apps/_executables.py

def execute(self) -> None:
    """Runs the provided callback."""
    self._callback()

AppBundle(*, app_plugin, container_plugin=EMPTY_PLUGIN, context=EMPTY_CONTEXT)

Bases: AppContainer

Brings together different types of containers to construct an executable application.

Use this class to defer the creation of an rats.apps.AppContainer instance in order to combine services with additional rats.apps.ContainerPlugin classes.

Create an instance by providing the [rats.apps.AppPlugin] type and any additional context.

Example

from rats import apps


class ExamplePlugin(apps.Container, apps.PluginMixin):
    @apps.service(apps.ServiceId[str]("some-value"))
    def _some_value() -> str:
        return "hello, world!"


class ExampleApplication(apps.AppContainer, apps.PluginMixin):
    def execute() -> None:
        print(self._app.get(apps.ServiceId[str]("some-value")))


if __name__ == "__main__":
    apps.run(apps.AppBundle(ExampleApplication, ExamplePlugin))

Parameters:

Name	Type	Description	Default
app_plugin	AppPlugin	the class reference to the application container.	required
container_plugin	ContainerPlugin	the class reference to an additional plugin container.	EMPTY_PLUGIN
context	Container	an optional plugin container to make part of the container tree.	EMPTY_CONTEXT

Source code in rats/apps/_app_containers.py

def __init__(
    self,
    *,
    app_plugin: AppPlugin,
    container_plugin: ContainerPlugin = EMPTY_PLUGIN,
    context: Container = EMPTY_CONTEXT,
):
    """
    Create an instance by providing the [rats.apps.AppPlugin] type and any additional context.

    Example:
        ```python
        from rats import apps


        class ExamplePlugin(apps.Container, apps.PluginMixin):
            @apps.service(apps.ServiceId[str]("some-value"))
            def _some_value() -> str:
                return "hello, world!"


        class ExampleApplication(apps.AppContainer, apps.PluginMixin):
            def execute() -> None:
                print(self._app.get(apps.ServiceId[str]("some-value")))


        if __name__ == "__main__":
            apps.run(apps.AppBundle(ExampleApplication, ExamplePlugin))
        ```

    Args:
        app_plugin: the class reference to the application container.
        container_plugin: the class reference to an additional plugin container.
        context: an optional plugin container to make part of the container tree.
    """
    self._app_plugin = app_plugin
    self._container_plugin = container_plugin
    self._context = context

execute()

Initializes a new [rats.apps.AppContainer] with the provided nodes before executing it.

Source code in rats/apps/_app_containers.py

def execute(self) -> None:
    """Initializes a new [rats.apps.AppContainer] with the provided nodes before executing it."""
    app, _ = self._get_or_create_containers()
    app.execute()

AppContainer

Bases: Container, Executable, Protocol

The combination of a rats.apps.Container an rats.apps.Executable.

execute()

The main application entry point.

Source code in rats/apps/_app_containers.py

def execute(self) -> None:
    """The main application entry point."""
    logger.warning(f"empty execute method in application: {self.__class__}")

CompositeContainer(*containers)

Bases: Container

Source code in rats/apps/_composite_container.py

def __init__(self, *containers: Container) -> None:
    self._containers = containers

get_namespaced_group(namespace, group_id)

Source code in rats/apps/_composite_container.py

def get_namespaced_group(
    self,
    namespace: str,
    group_id: ServiceId[T_ServiceType],
) -> Iterator[T_ServiceType]:
    for container in self._containers:
        yield from container.get_namespaced_group(namespace, group_id)

CompositePlugin(*plugins)

Similar to rats.apps.CompositeContainer but takes a list of plugin container types.

Example

from rats import apps
from rats_e2e.apps import inputs


class ExamplePlugin1(apps.Container, apps.PluginMixin):
    pass


class ExamplePlugin2(apps.Container, apps.PluginMixin):
    pass


apps.run(
    apps.AppBundle(
        app_plugin=inputs.Application,
        container_plugin=apps.CompositePlugin(
            ExamplePlugin1,
            ExamplePlugin2,
        ),
    )
)

Source code in rats/apps/_app_containers.py

def __init__(self, *plugins: ContainerPlugin) -> None:
    self._plugins = plugins

__call__(app)

Source code in rats/apps/_app_containers.py

def __call__(self, app: Container) -> Container:
    return CompositeContainer(*[plugin(app) for plugin in self._plugins])

Container

Bases: Protocol

Main interface for service containers.

The default methods in this protocol attempt to find service providers that have been annotated.

Example

.. code-block:: python

from rats import apps


class MyStorageClient:
    def save(self, data: str) -> None:
        print(f"Saving data: {data}")


class MyPluginServices:
    STORAGE_CLIENT = ServiceId[MyStorageClient]("storage-client")


class MyPluginContainer(apps.Container):
    @apps.service(MyPluginServices.STORAGE_CLIENT)
    def _storage_client() -> MyStorageClient:
        return MyStorageClient()


container = MyPluginContainer()
storage_client = container.get(MyPluginServices.STORAGE_CLIENT)
storage_client.save("Hello, world!")

get(service_id)

Retrieve a service instance by its id.

Source code in rats/apps/_container.py

def get(self, service_id: ServiceId[T_ServiceType]) -> T_ServiceType:
    """Retrieve a service instance by its id."""
    services = list(self.get_namespaced_group(ProviderNamespaces.SERVICES, service_id))
    if len(services) == 0:
        services.extend(
            list(self.get_namespaced_group(ProviderNamespaces.FALLBACK_SERVICES, service_id)),
        )

    if len(services) > 1:
        raise DuplicateServiceError(service_id)
    elif len(services) == 0:
        raise ServiceNotFoundError(service_id)
    else:
        return services[0]

get_group(group_id)

Retrieve a service group by its id.

Source code in rats/apps/_container.py

def get_group(
    self,
    group_id: ServiceId[T_ServiceType],
) -> Iterator[T_ServiceType]:
    """Retrieve a service group by its id."""
    if not self.has_namespace(ProviderNamespaces.GROUPS, group_id):
        # groups are expected to return iterable services
        # TODO: we need to clean up the meaning of groups and services somehow
        for i in self.get_namespaced_group(ProviderNamespaces.FALLBACK_GROUPS, group_id):
            yield from cast(Iterator[T_ServiceType], i)

    for i in self.get_namespaced_group(ProviderNamespaces.GROUPS, group_id):
        yield from cast(Iterator[T_ServiceType], i)

get_namespaced_group(namespace, group_id)

Retrieve a service group by its id, within a given service namespace.

Source code in rats/apps/_container.py

def get_namespaced_group(
    self,
    namespace: str,
    group_id: ServiceId[T_ServiceType],
) -> Iterator[T_ServiceType]:
    """Retrieve a service group by its id, within a given service namespace."""
    yield from _get_cached_services_for_group(self, namespace, group_id)

    for subcontainer in _get_subcontainers(self):
        yield from subcontainer.get_namespaced_group(namespace, group_id)

has(service_id)

Check if a service is provided by this container.

Example

.. code-block:: python

if not container.has(MyPluginServices.STORAGE_CLIENT):
    print("Did you forget to configure a storage client?")

Source code in rats/apps/_container.py

def has(self, service_id: ServiceId[T_ServiceType]) -> bool:
    """
    Check if a service is provided by this container.

    Example:
        .. code-block:: python

            if not container.has(MyPluginServices.STORAGE_CLIENT):
                print("Did you forget to configure a storage client?")
    """
    try:
        return self.get(service_id) is not None
    except ServiceNotFoundError:
        return False

has_group(group_id)

Check if a service group has at least one provider in the container.

Source code in rats/apps/_container.py

def has_group(self, group_id: ServiceId[T_ServiceType]) -> bool:
    """Check if a service group has at least one provider in the container."""
    try:
        return next(self.get_group(group_id)) is not None
    except StopIteration:
        return False

has_namespace(namespace, group_id)

Source code in rats/apps/_container.py

def has_namespace(self, namespace: str, group_id: ServiceId[T_ServiceType]) -> bool:
    try:
        return next(self.get_namespaced_group(namespace, group_id)) is not None
    except StopIteration:
        return False

DuplicateServiceError(service_id)

Bases: RuntimeError, Generic[T_ServiceType]

Source code in rats/apps/_container.py

def __init__(self, service_id: ServiceId[T_ServiceType]) -> None:
    super().__init__(f"Service id provided multiple times: {service_id}")
    self.service_id = service_id

service_id = service_id instance-attribute

Executable

Bases: Protocol

An interface for an executable object.

One of the lowest level abstractions in the rats-apps library, executables are meant to be easy to run from anywhere, with limited knowledge of the implementation details of the object, by ensuring that the object has an execute method with no arguments.

execute() abstractmethod

Execute the application.

Source code in rats/apps/_executables.py

@abstractmethod
def execute(self) -> None:
    """Execute the application."""

GroupProvider

Bases: Protocol[Tco_ServiceType]

__call__() abstractmethod

Return the group instances.

Source code in rats/apps/_container.py

@abstractmethod
def __call__(self) -> Iterator[Tco_ServiceType]:
    """Return the group instances."""

NullRuntime(msg)

Bases: Runtime

Source code in rats/apps/_runtimes.py

def __init__(self, msg: str) -> None:
    self._msg = msg

execute(*exe_ids)

Source code in rats/apps/_runtimes.py

def execute(self, *exe_ids: ServiceId[T_ExecutableType]) -> None:
    logger.error(self._msg)
    raise NotImplementedError(f"NullRuntime cannot execute ids: {exe_ids}")

execute_callable(*callables)

Source code in rats/apps/_runtimes.py

def execute_callable(self, *callables: Callable[[], None]) -> None:
    raise NotImplementedError(f"NullRuntime cannot execute callables: {callables}")

execute_group(*exe_group_ids)

Source code in rats/apps/_runtimes.py

def execute_group(self, *exe_group_ids: ServiceId[T_ExecutableType]) -> None:
    raise NotImplementedError(f"NullRuntime cannot execute groups: {exe_group_ids}")

PluginContainers(app, group, *names)

Bases: Container

A container that loads plugins using importlib.metadata.entry_points.

When looking for groups, the container loads the specified entry_points and defers the lookups to the plugins. Plugin containers are expected to be Callable[[Container], Container] objects, where the input container is typically the root application container.

TODO: How do we better specify the API for plugins without relying on documentation?

Source code in rats/apps/_plugin_container.py

def __init__(self, app: Container, group: str, *names: str) -> None:
    self._app = app
    self._group = group
    self._names = names

get_namespaced_group(namespace, group_id)

Source code in rats/apps/_plugin_container.py

def get_namespaced_group(
    self,
    namespace: str,
    group_id: ServiceId[T_ServiceType],
) -> Iterator[T_ServiceType]:
    for container in self._load_containers():
        yield from container.get_namespaced_group(namespace, group_id)

PluginMixin(app)

Mix into your [Container][] classes to add our default constructor.

This mixin adds a common constructor to a Container in order to quickly create types that are compatible with functions asking for AppPlugin and ContainerPlugin arguments.

Warning

Avoid using mixins as an input type to your functions, because we don't want to restrict others to containers with a private _app property. Instead, use this as a shortcut to some commonly used implementation details.

Example

 from rats import apps

 class ExampleApplication(apps.AppContainer, apps.PluginMixin):

    def execute() -> None:
        print("hello, world!")


if __name__ == "__main__":
    apps.run_plugin(ExampleApplication)

Source code in rats/apps/_app_containers.py

def __init__(self, app: Container) -> None:
    self._app = app

Provider

Bases: Protocol[Tco_ServiceType]

__call__() abstractmethod

Return the service instance.

Source code in rats/apps/_container.py

@abstractmethod
def __call__(self) -> Tco_ServiceType:
    """Return the service instance."""

ProviderNamespaces

CONTAINERS = 'containers' class-attribute instance-attribute

FALLBACK_GROUPS = 'fallback-groups' class-attribute instance-attribute

FALLBACK_SERVICES = 'fallback-services' class-attribute instance-attribute

GROUPS = 'groups' class-attribute instance-attribute

SERVICES = 'services' class-attribute instance-attribute

Runtime

Bases: Protocol

execute(*exe_ids) abstractmethod

Execute a list of executables sequentially.

Source code in rats/apps/_runtimes.py

@abstractmethod
def execute(self, *exe_ids: ServiceId[T_ExecutableType]) -> None:
    """Execute a list of executables sequentially."""

execute_group(*exe_group_ids) abstractmethod

Execute one or more groups of executables sequentially.

Although each group is expected to be executed sequentially, the groups themselves are not executed in a deterministic order. Runtime implementations are free to execute groups in parallel or in any order that is convenient.

Source code in rats/apps/_runtimes.py

@abstractmethod
def execute_group(self, *exe_group_ids: ServiceId[T_ExecutableType]) -> None:
    """
    Execute one or more groups of executables sequentially.

    Although each group is expected to be executed sequentially, the groups themselves are not
    executed in a deterministic order. Runtime implementations are free to execute groups in
    parallel or in any order that is convenient.
    """

ServiceId

Bases: NamedTuple, Generic[T_ServiceType]

name instance-attribute

ServiceNotFoundError(service_id)

Bases: RuntimeError, Generic[T_ServiceType]

Source code in rats/apps/_container.py

def __init__(self, service_id: ServiceId[T_ServiceType]) -> None:
    super().__init__(f"Service id not found: {service_id}")
    self.service_id = service_id

service_id = service_id instance-attribute

StandardRuntime(app)

Bases: Runtime

A simple runtime that executes sequentially and in a single thread.

Source code in rats/apps/_runtimes.py

def __init__(self, app: Container) -> None:
    self._app = app

execute(*exe_ids)

Source code in rats/apps/_runtimes.py

def execute(self, *exe_ids: ServiceId[T_ExecutableType]) -> None:
    for exe_id in exe_ids:
        self._app.get(exe_id).execute()

execute_group(*exe_group_ids)

Source code in rats/apps/_runtimes.py

def execute_group(self, *exe_group_ids: ServiceId[T_ExecutableType]) -> None:
    for exe_group_id in exe_group_ids:
        for exe in self._app.get_group(exe_group_id):
            exe.execute()

StaticContainer(*providers)

Bases: Container

Source code in rats/apps/_static_container.py

def __init__(self, *providers: StaticProvider[Any]) -> None:
    self._providers = providers

get_namespaced_group(namespace, group_id)

Source code in rats/apps/_static_container.py

def get_namespaced_group(
    self,
    namespace: str,
    group_id: ServiceId[T_ServiceType],
) -> Iterator[T_ServiceType]:
    for provider in self._providers:
        if provider.namespace == namespace and provider.service_id == group_id:
            yield provider.call()

StaticProvider(namespace, service_id, call) dataclass

Bases: Generic[T_ServiceType]

call instance-attribute

namespace instance-attribute

service_id instance-attribute

autoid(method)

Get a service id for a method.

The service id is constructed from the module, class and method name. It should be identical regardless of whether the method is bound or not, and regardless of the instance it is bound to.

The service type is the return type of the method.

Source code in rats/apps/_annotations.py

def autoid(method: Callable[..., T_ServiceType]) -> ServiceId[T_ServiceType]:
    """
    Get a service id for a method.

    The service id is constructed from the module, class and method name.  It should be identical
    regardless of whether the method is bound or not, and regardless of the instance it is bound
    to.

    The service type is the return type of the method.
    """
    service_name = _get_method_service_id_name(method)
    return ServiceId[T_ServiceType](service_name)

autoid_factory_service(method)

A decorator to create a factory service, with an automatically generated service id.

Decorate a method that takes any number of arguments and returns an object. The resulting service will be that factory - taking the same arguments and returning a new object each time.

Source code in rats/apps/_annotations.py

def autoid_factory_service(
    method: Callable[Concatenate[T_Container, P], R],
) -> Callable[[T_Container], Callable[P, R]]:
    """
    A decorator to create a factory service, with an automatically generated service id.

    Decorate a method that takes any number of arguments and returns an object.  The resulting
    service will be that factory - taking the same arguments and returning a new object each time.
    """
    new_method = _factory_to_factory_provider(method)
    return autoid_service(new_method)

autoid_service(fn)

Source code in rats/apps/_annotations.py

def autoid_service(fn: Callable[P, T_ServiceType]) -> Callable[P, T_ServiceType]:
    _service_id = autoid(fn)
    return annotations.annotation(ProviderNamespaces.SERVICES, cast(NamedTuple, _service_id))(fn)

autoscope(cls)

Decorator to automatically scope ServiceId attributes in a class.

Source code in rats/apps/_scoping.py

def autoscope(cls: T) -> T:
    """Decorator to automatically scope ServiceId attributes in a class."""

    def _wrap(func: Callable[P, Any]) -> Callable[P, Any]:
        def _wrapper(*args: P.args, **kwargs: P.kwargs) -> ServiceId[Any]:
            result = func(*args, **kwargs)
            if not isinstance(result, ServiceId):
                return result

            return ServiceId[Any](scope_service_name(cls.__module__, cls.__name__, result.name))

        return _wrapper

    props = [prop for prop in dir(cls) if not prop.startswith("_")]

    for prop_name in props:
        non_ns = getattr(cls, prop_name)

        if isinstance(non_ns, FunctionType):
            setattr(cls, prop_name, _wrap(non_ns))
        else:
            if not isinstance(non_ns, ServiceId):
                continue

            prop = ServiceId[Any](scope_service_name(cls.__module__, cls.__name__, non_ns.name))
            setattr(cls, prop_name, prop)

    return cls

container(group_id=DEFAULT_CONTAINER_GROUP)

Source code in rats/apps/_container.py

def container(
    group_id: ServiceId[T_ServiceType] = DEFAULT_CONTAINER_GROUP,
) -> Callable[[Callable[P, T_ServiceType]], Callable[P, T_ServiceType]]:
    return annotations.annotation(ProviderNamespaces.CONTAINERS, cast(NamedTuple, group_id))

fallback_group(group_id)

A fallback group gets used if no group is defined.

Source code in rats/apps/_annotations.py

def fallback_group(
    group_id: ServiceId[T_ServiceType],
) -> Callable[[Callable[P, Iterator[T_ServiceType]]], Callable[P, Iterator[T_ServiceType]]]:
    """A fallback group gets used if no group is defined."""
    return annotations.annotation(
        ProviderNamespaces.FALLBACK_GROUPS,
        cast(NamedTuple, group_id),
    )

fallback_service(service_id)

A fallback service gets used if no service is defined.

Source code in rats/apps/_annotations.py

def fallback_service(
    service_id: ServiceId[T_ServiceType],
) -> Callable[[Callable[P, T_ServiceType]], Callable[P, T_ServiceType]]:
    """A fallback service gets used if no service is defined."""
    return annotations.annotation(
        ProviderNamespaces.FALLBACK_SERVICES,
        cast(NamedTuple, service_id),
    )

group(group_id)

A group is a collection of services.

Source code in rats/apps/_annotations.py

def group(
    group_id: ServiceId[T_ServiceType],
) -> Callable[[Callable[P, Iterator[T_ServiceType]]], Callable[P, Iterator[T_ServiceType]]]:
    """A group is a collection of services."""
    return annotations.annotation(ProviderNamespaces.GROUPS, cast(NamedTuple, group_id))

run(*apps)

Shortcut for running a list of apps.

Source code in rats/apps/_mains.py

def run(*apps: Executable) -> None:
    """Shortcut for running a list of apps."""
    for app in apps:
        app.execute()

run_plugin(*app_plugins)

Shortcut to create and execute instances of apps.AppPlugin.

This function is most commonly used in a console_script function main() entry point.

Example

from rats import apps


class Application(apps.AppContainer, apps.AppPlugin):
    def execute(self) -> None:
        print("hello, world")


def main() -> None:
    apps.run_plugin(Application)


if __name__ == "__main__":
    main()

Parameters:

Name	Type	Description	Default
*app_plugins	AppPlugin	one or more class types to be instantiated and executed.	()

Source code in rats/apps/_mains.py

def run_plugin(*app_plugins: AppPlugin) -> None:
    """
    Shortcut to create and execute instances of `apps.AppPlugin`.

    This function is most commonly used in a `console_script` function `main()` entry point.

    Example:
        ```python
        from rats import apps


        class Application(apps.AppContainer, apps.AppPlugin):
            def execute(self) -> None:
                print("hello, world")


        def main() -> None:
            apps.run_plugin(Application)


        if __name__ == "__main__":
            main()
        ```

    Args:
        *app_plugins: one or more class types to be instantiated and executed.
    """
    run(*[AppBundle(app_plugin=plugin) for plugin in app_plugins])

service(service_id)

A service is anything you would create instances of?

Source code in rats/apps/_annotations.py

def service(
    service_id: ServiceId[T_ServiceType],
) -> Callable[[Callable[P, T_ServiceType]], Callable[P, T_ServiceType]]:
    """A service is anything you would create instances of?"""
    return annotations.annotation(ProviderNamespaces.SERVICES, cast(NamedTuple, service_id))

static_group(group_id, provider)

Factory function for a StaticProvider instance for ProviderNamespaces.GROUPS.

Warning

Unlike group providers in a container, the provider function argument here should return a single instance of the service group.

Parameters:

Name	Type	Description	Default
group_id	ServiceId[T_ServiceType]	the identifier for the provided service group.	required
provider	Provider[T_ServiceType]	a callable that returns an instance of T_ServiceType.	required

Returns: StaticProvider instance for the provided group_id.

Source code in rats/apps/_static_container.py

def static_group(
    group_id: ServiceId[T_ServiceType],
    provider: Provider[T_ServiceType],
) -> StaticProvider[T_ServiceType]:
    """
    Factory function for a `StaticProvider` instance for `ProviderNamespaces.GROUPS`.

    !!! warning
        Unlike group providers in a container, the provider function argument here should return
        a single instance of the service group.

    Args:
        group_id: the identifier for the provided service group.
        provider: a callable that returns an instance of T_ServiceType.

    Returns: StaticProvider instance for the provided group_id.
    """
    return StaticProvider(ProviderNamespaces.GROUPS, group_id, provider)

static_service(service_id, provider)

Factory function for a StaticProvider instance for ProviderNamespaces.SERVICES.

Parameters:

Name	Type	Description	Default
service_id	ServiceId[T_ServiceType]	the identifier for the provided service.	required
provider	Provider[T_ServiceType]	a callable that returns an instance of T_ServiceType.	required

Returns: StaticProvider instance for the provided service_id.

Source code in rats/apps/_static_container.py

def static_service(
    service_id: ServiceId[T_ServiceType],
    provider: Provider[T_ServiceType],
) -> StaticProvider[T_ServiceType]:
    """
    Factory function for a `StaticProvider` instance for `ProviderNamespaces.SERVICES`.

    Args:
        service_id: the identifier for the provided service.
        provider: a callable that returns an instance of T_ServiceType.

    Returns: StaticProvider instance for the provided service_id.
    """
    return StaticProvider(ProviderNamespaces.SERVICES, service_id, provider)

Examples

The rats_e2e.apps module has a handful of heavily documented tests that serve as good examples for different types of applications. You can run the cli command at the root of the module with python -m rats_e2e.apps or interact with each example directly.

We try to make each example fully executable by defining an entry point main(), the function we would reference to register a script in pyproject.toml; the __main__.py file makes the modules executable directly, like python -m rats_e2e.apps.[name]; if the example is of an application, a ._app.py file will contain the Application class and any relevant service ids; and if the example contains a plugin container–meant to be loaded by another application–the relevant PluginContainer and service ids will be found in a ._plugin.py file.

rats_e2e.apps.minimal

A minimal executable application example.

You can run this example by using the rats_e2e.apps.minimal.Application class within python, or directly through a terminal.

python ~/code

from rats import apps
from rats_e2e.apps import minimal

apps.run_plugin(minimal.Application)

python -m rats_e2e.apps.minimal

__all__ = ['Application', 'main'] module-attribute

Application(app)

Bases: apps.AppContainer, apps.PluginMixin

Prints a handful of random values to stdout.

We can run this application with apps.run_plugin(minimal.Application) or directly in the terminal with python -m rats_e2e.apps.minimal. Use the RATS_E2E_NUM_VALUES environment variable to alter the number of values to print.

Source code in rats/apps/_app_containers.py

def __init__(self, app: Container) -> None:
    self._app = app

execute()

The main entry point to the application.

Source code in rats_e2e/apps/minimal/_app.py

def execute(self) -> None:
    """The main entry point to the application."""
    logger.info("running minimal application example")
    for _x in range(int(os.environ.get("RATS_E2E_NUM_VALUES", 5))):
        print(uuid4())

main()

Entry point function to register this application as a script.

Source code in rats_e2e/apps/minimal/_app.py

def main() -> None:
    """Entry point function to register this application as a script."""
    apps.run_plugin(Application)

rats_e2e.apps.inputs

Application using services expected to be provided externally.

Instead of using the RATS_E2E_NUM_VALUES environment variable, this application introduces a small configuration object rats_e2e.apps.inputs.AppInput that we can specify when running things.

from rats import apps
from rats_e2e.apps import inputs


class AppContext(apps.Container, apps.PluginMixin):
    @apps.service(inputs.AppServices.INPUT)
    def _input(self) -> inputs.AppInput:
        return inputs.AppInput(randint(0, 5))


ctx = ExampleContextContainer(apps.AppContext())
apps.run(apps.AppBundle(app_plugin=inputs.Application, context=ctx))

__all__ = ['AppInput', 'AppServices', 'Application', 'main'] module-attribute

AppInput

Bases: NamedTuple

A small data structure to provide the needed configuration for our application.

num_rows instance-attribute

The number of values we want printed by the rats_e2e.apps.inputs.Application class.

AppServices

INPUT = apps.ServiceId[AppInput]('input') class-attribute instance-attribute

Application(app)

Bases: apps.AppContainer, apps.PluginMixin

Prints a handful of random values to stdout.

We can run this application with apps.run_plugin(minimal.Application) or directly in the terminal with python -m rats_e2e.apps.inputs. However, the library api allows the addition of a rats.apps.Container with a service used as configuration.

Source code in rats/apps/_app_containers.py

def __init__(self, app: Container) -> None:
    self._app = app

execute()

The main entry point to the application.

Source code in rats_e2e/apps/inputs/_app.py

def execute(self) -> None:
    """The main entry point to the application."""
    app_input = self._app.get(AppServices.INPUT)
    logger.info(f"running inputs application example with input: {app_input}")
    for _x in range(app_input.num_rows):
        print(uuid4())

main()

Source code in rats_e2e/apps/inputs/_app.py

def main() -> None:
    apps.run_plugin(Application)