Skip to content

rats.runtime

Run rats applications, providing external context.

The provided rats-runtime cli allows the registration and execution of rats.apps.AppContainer applications. Register to the python plugin group called rats.runtime.apps, and list available plugins with rats-runtime list.

pyproject.toml
[project.entry-points."rats.runtime.apps"]
"rats_e2e.runtime" = "rats_e2e.runtime:Application"

We register an example application you can run with rats-runtime run rats_e2e.runtime. The name of the registered entry point in your pyproject.toml will map to the name provided to rats-runtime run.

$ rats-runtime run rats_e2e.runtime
hello, world!
looking for any registered context: rats_e2e.runtime._app:AppServices[example-data]

Our example application looks for any context services, which can be provided to the run command.

$ rats-runtime run         rats_e2e.runtime         --context-file src/rats_resources/runtime/example-context.yaml
hello, world!
looking for any registered context: rats_e2e.runtime._app:AppServices[example-data]
found example data element: ExampleData(id='111', thing_a='111', thing_b='111')
found example data element: ExampleData(id='222', thing_a='222', thing_b='222')

items:
  - service_id: ["rats_e2e.runtime._app:AppServices[example-data]"]
    values:
      - id: "111"
        thing_a: "111"
        thing_b: "111"
      - id: "222"
        thing_a: "222"
        thing_b: "222"

from collections.abc import Iterator

from rats import apps, runtime

from ._data import ExampleData


@apps.autoscope
class AppServices:
    EXAMPLE_DATA = apps.ServiceId[ExampleData]("example-data")


class Application(apps.AppContainer, apps.PluginMixin):
    def execute(self) -> None:
        print("hello, world!")
        print(f"looking for any registered context: {AppServices.EXAMPLE_DATA.name}")
        for data in self._app.get_group(AppServices.EXAMPLE_DATA):
            print(f"found example data element: {data}")

    @apps.fallback_group(AppServices.EXAMPLE_DATA)
    def _data_from_context(self) -> Iterator[ExampleData]:
        collection = self._app.get(runtime.AppServices.CONTEXT)
        yield from collection.decoded_values(ExampleData, AppServices.EXAMPLE_DATA)

from dataclasses import dataclass


@dataclass(frozen=True)
class ExampleData:
    id: str
    thing_a: str
    thing_b: str

Info

Most users won't typically use rats-runtime directly, but this interface is used by other abstractions to execute applications remotely, like the rats.aml module, adding the ability to execute applications within an azure ml job.

__all__ = ['AppServices', 'Application', 'DuplicateRequestError', 'Request', 'RequestNotFoundError', 'main'] module-attribute

Application(app)

Bases: apps.AppContainer, cli.Container, apps.PluginMixin

The rats-runtime cli application.

$ rats-runtime --help
Usage: rats-runtime [OPTIONS] COMMAND [ARGS]...

Options:
  --help  Show this message and exit.

Commands:
  list  List all the exes and groups that are configured to run with `rats-
        runtime run`.
  run   Run one or more apps, adding an optional additional context.
Source code in rats/apps/_app_containers.py 
90
91
def __init__(self, app: Container) -> None:
    self._app = app

execute()

Runs the rats-runtime cli.

Source code in rats/runtime/_app.py 
70
71
72
73
74
75
76
77
78
79
def execute(self) -> None:
    """Runs the `rats-runtime` cli."""
    argv = self._app.get(cli.PluginConfigs.ARGV)
    cli.create_group(click.Group("rats-runtime"), self).main(
        args=argv[1:],
        prog_name=Path(argv[0]).name,
        auto_envvar_prefix="RATS_RUNTIME",
        # don't end the process
        standalone_mode=False,
    )

AppServices

Services used by the rats.runtime.Application cli command.

CONTEXT = apps.ServiceId[app_context.Collection[Any]]('app-ctx-collection.config') class-attribute instance-attribute

rats.app_context.Collection available in the application run using rats-runtime run.

from rats import apps, runtime


class Application(apps.AppContainer, apps.PluginMixin):
    def execute(self) -> None:
        context = self._app.get(runtime.AppConfigs.CONTEXT)
        print("loaded context:")
        for item in context_collection.items:
            print(f"{item.service_id} -> {item.values}")

REQUEST = apps.ServiceId[Request]('runtime-details-client') class-attribute instance-attribute

The request information, available after the call to rats.runtime.Application.execute.

DuplicateRequestError()

Bases: RuntimeError

Source code in rats/runtime/_request.py 
22
23
def __init__(self) -> None:
    super().__init__("request already executed.")

Request

Bases: NamedTuple

app_ids instance-attribute

The app ids being run by rats.runtime.Application.

context instance-attribute

The rats.app_context.Collection that was submitted and made available to executed apps.

RequestNotFoundError()

Bases: RuntimeError

Thrown if rats.runtime.AppServices.REQUEST is accessed without the application being run.

Source code in rats/runtime/_request.py 
17
18
def __init__(self) -> None:
    super().__init__("no running runtime request found.")

main()

Main entry-point to the rats-runtime cli command.

Source code in rats/runtime/_app.py 
153
154
155
156
157
158
159
160
def main() -> None:
    """Main entry-point to the `rats-runtime` cli command."""
    try:
        apps.run_plugin(logs.ConfigureApplication)
        apps.run_plugin(Application)
    except click.exceptions.ClickException as e:
        e.show()
        sys.exit(e.exit_code)