Skip to content

rats.app_context

Allows marshaling of simple value objects across processes and machines.

The rats.app_context.Collection and rats.app_context.Context classes are the building blocks for creating a collection of services that can be serialized to json and shared across applications. In order to be easily serializable, all service values managed by this module must be instances of dataclasses.dataclass, must be immutable, and must be made up of simple value types.

Example

If we have an example data structure, MySimpleData, we can store instances of it into a rats.app_context.Context object by assigning them a service id. Every service id in a context maps to zero or more instances of our data structure. Zero or more context objects can be used to create rats.app_context.Collection instances, which can be serialized with the rats.app_context.dumps and rats.app_context.loads functions.

from dataclasses import dataclass
from rats import apps, app_context


@dataclass(frozen=True)
class MySimpleData:
    blob_path: tuple[str, str, str]
    offset: int


data1 = MySimpleData(
    blob_path=("accountname", "containername", "/some/blob/path"),
    offset=3,
)

data2 = MySimpleData(
    blob_path=("accountname", "containername", "/another/blob/path"),
    offset=3,
)

service_id = apps.ServiceId[MySimpleData]("example-service")

ctx = app_context.Context[MySimpleData].make(service_id, data1, data2)
collection = app_context.Collection.make(ctx)

# the encoded json string can be moved between machines through files, env variables, etc.
json_encoded = app_context.dumps(collection)
# on the remote process, we can rebuild the original collection object
original_collection = app_context.loads(json_encoded)
# we can retrieve the json-encoded contexts
values = original_collection.values(service_id)
# and we can decode the values back into the `MySimpleData` instances
simple_data_instances = original_collection.decoded_values(MySimpleData, service_id)

EMPTY_COLLECTION = Collection[Any].empty() module-attribute

Empty collection constant usable as default method values.

ContextValue = dict[str, Any] module-attribute

Values found in rats.app_context.Context in their simple value type representations.

T_ContextType = TypeVar('T_ContextType') module-attribute

Generic type for context value objects.

Warning

Python lacks the ability to bind generic values to require them to be a dataclasses.dataclass instance, but all rats.app_context.T_ContextType instances must be dataclasses in order for them to serialize and deserialize. This is mostly a leaky private detail, and we hope to remove this requirement in the future.

__all__ = ['EMPTY_COLLECTION', 'Collection', 'Context', 'ContextValue', 'GroupContainer', 'ServiceContainer', 'T_ContextType', 'dumps', 'loads'] module-attribute

Collection(items) dataclass

Bases: Generic[T_ContextType]

Collection of context objects linking service ids to serializable data structures.

It is up to the user of this class to provide the correct mapping between services and the constructor for those services.

items instance-attribute

Access to the raw rats.app_context.Context instances in the collection.

merge(*collections) staticmethod

Merge multiple collections into a new, combined instance.

Parameters:

Name Type Description Default
*collections Collection[T_ContextType]

zero or more collections to merge into one.

 ()
Source code in rats/app_context/_collection.py 
30
31
32
33
34
35
36
37
38
39
40
@staticmethod
def merge(*collections: Collection[T_ContextType]) -> Collection[T_ContextType]:
    """
    Merge multiple collections into a new, combined instance.

    Args:
        *collections: zero or more collections to merge into one.
    """
    return Collection[T_ContextType].make(
        *[ctx for collection in collections for ctx in collection.items]
    )

empty() staticmethod

A shortcut to retrieving an empty collection.

Source code in rats/app_context/_collection.py 
42
43
44
45
@staticmethod
def empty() -> Collection[T_ContextType]:
    """A shortcut to retrieving an empty collection."""
    return Collection(items=())

make(*items) staticmethod

Convenience function to create a collection from a list of contexts.

Parameters:

Name Type Description Default
*items Context[T_ContextType]

zero or more context objects to turn into a collection.

 ()
Source code in rats/app_context/_collection.py 
47
48
49
50
51
52
53
54
55
@staticmethod
def make(*items: Context[T_ContextType]) -> Collection[T_ContextType]:
    """
    Convenience function to create a collection from a list of contexts.

    Args:
        *items: zero or more context objects to turn into a collection.
    """
    return Collection(items=items)

service_ids()

Get the list of service items found in the collection, across all context instances.

Source code in rats/app_context/_collection.py 
57
58
59
def service_ids(self) -> set[apps.ServiceId[T_ContextType]]:
    """Get the list of service items found in the collection, across all context instances."""
    return {item.service_id for item in self.items}

decoded_values(cls, service_id)

Given a reference to a dataclass type, retrieves and builds instances matching a service id.

Info

The dataclass used in the two communicating processes does not need to be the same, but we expect the constructor arguments to be compatible.

Parameters:

Name Type Description Default
cls type[T_ContextType]

the type all selected context objects will be cast into before returning.

 required
service_id apps.ServiceId[T_ContextType]

the selector for the expected context objects.

 required
Source code in rats/app_context/_collection.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def decoded_values(
    self,
    cls: type[T_ContextType],
    service_id: apps.ServiceId[T_ContextType],
) -> tuple[T_ContextType, ...]:
    """
    Given a reference to a dataclass type, retrieves and builds instances matching a service id.

    !!! info
        The dataclass used in the two communicating processes does not need to be the same, but
        we expect the constructor arguments to be compatible.

    Args:
        cls: the type all selected context objects will be cast into before returning.
        service_id: the selector for the expected context objects.
    """
    return tuple(dataclass_wizard.fromlist(cls, list(self.values(service_id))))

values(service_id)

Retrieves the raw data structures matching a service id.

The values returned here have been encoded into simple dictionaries and are ready to be serialized and transferred across machines. See the rats.app_context.Collection.decoded_values method to retrieve context values that have been turned back into the desired dataclass objects.

Parameters:

Name Type Description Default
service_id apps.ServiceId[T_ContextType]

the selector for the expected context objects.

 required
Source code in rats/app_context/_collection.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def values(self, service_id: apps.ServiceId[T_ContextType]) -> tuple[ContextValue, ...]:
    """
    Retrieves the raw data structures matching a service id.

    The values returned here have been encoded into simple dictionaries and are ready to be
    serialized and transferred across machines. See the
    [rats.app_context.Collection.decoded_values][] method to retrieve context values that have
    been turned back into the desired dataclass objects.

    Args:
        service_id: the selector for the expected context objects.
    """
    results: list[ContextValue] = []
    for item in self.with_id(service_id).items:
        for value in item.values:
            results.append(value)

    return tuple(results)

add(*items)

Creates a new Collection with the provided items added to the current ones.

Parameters:

Name Type Description Default
*items Context[T_ContextType]

the context items to be added in the new collection.

 ()
Source code in rats/app_context/_collection.py 
 98
 99
100
101
102
103
104
105
106
107
108
def add(self, *items: Context[T_ContextType]) -> Collection[T_ContextType]:
    """
    Creates a new Collection with the provided items added to the current ones.

    Args:
        *items: the context items to be added in the new collection.
    """
    return Collection[T_ContextType].make(
        *self.items,
        *items,
    )

with_id(*service_ids)

Filter out context items not matching the provided service ids and return a new Collection.

Parameters:

Name Type Description Default
*service_ids apps.ServiceId[T_ContextType]

the selector for the returned context collection.

 ()
Source code in rats/app_context/_collection.py 
110
111
112
113
114
115
116
117
118
119
def with_id(self, *service_ids: apps.ServiceId[T_ContextType]) -> Collection[T_ContextType]:
    """
    Filter out context items not matching the provided service ids and return a new Collection.

    Args:
        *service_ids: the selector for the returned context collection.
    """
    return Collection[T_ContextType](
        items=tuple(item for item in self.items if item.service_id in service_ids),
    )

GroupContainer(cls, collection)

Bases: apps.Container, Generic[T_ContextType]

A rats.apps.Container that provides services from a rats.app_context.Collection.

Source code in rats/app_context/_container.py 
66
67
68
69
70
71
72
def __init__(
    self,
    cls: type[T_ContextType],
    collection: Collection[T_ContextType],
) -> None:
    self._cls = cls
    self._collection = collection

ServiceContainer(cls, namespace, collection)

Bases: apps.Container, Generic[T_ContextType]

A rats.apps.Container that provides services from a rats.app_context.Collection.

Source code in rats/app_context/_container.py 
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    cls: type[T_ContextType],
    namespace: str,
    collection: Collection[T_ContextType],
) -> None:
    self._cls = cls
    self._namespace = namespace
    self._collection = collection

Context(service_id, values) dataclass

Bases: dataclass_wizard.JSONSerializable, Generic[T_ContextType]

An easily serializable class to hold one or more values tied to a service id.

We can use these instances to share simple value types across a cluster of machines, and integrate them with rats.apps.Container to make the business logic unaware of any complexity in here.

service_id instance-attribute

The rats.apps.ServiceId the remote app will be able to use to retrieve the stored values.

values instance-attribute

The encoded value object, ready to be turned into json for marshaling.

make(service_id, *contexts) staticmethod

Convert a dataclass into a context object tied to the provided service id.

from dataclass import dataclass


@dataclass(frozen=True)
class MySimpleData:
    blob_path: Tuple[str, str, str]
    offset: int


service_id = apps.ServiceId[MySimpleData]("example-service")
ctx = app_context.Context[MySimpleData].make(
    service_id,
    MySimpleData(
        blob_path=("accountname", "containername", "/some/blob/path"),
    ),
)

Parameters:

Name Type Description Default
service_id apps.ServiceId[T_ContextType]

The service id the context values will be retrievable as.

 required
*contexts T_ContextType

Zero or more rats.app_context.T_ContextType instances.

 ()
Source code in rats/app_context/_context.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@staticmethod
def make(
    service_id: apps.ServiceId[T_ContextType],
    *contexts: T_ContextType,
) -> Context[T_ContextType]:
    """
    Convert a dataclass into a context object tied to the provided service id.

    ```python
    from dataclass import dataclass


    @dataclass(frozen=True)
    class MySimpleData:
        blob_path: Tuple[str, str, str]
        offset: int


    service_id = apps.ServiceId[MySimpleData]("example-service")
    ctx = app_context.Context[MySimpleData].make(
        service_id,
        MySimpleData(
            blob_path=("accountname", "containername", "/some/blob/path"),
        ),
    )
    ```

    Args:
        service_id: The service id the context values will be retrievable as.
        *contexts: Zero or more [rats.app_context.T_ContextType][] instances.
    """
    if len(contexts) > 0:
        cls = contexts[0].__class__
        dataclass_wizard.DumpMeta(key_transform="NONE").bind_to(cls)  # type: ignore[reportUnknownMemberType]
    return Context(
        service_id,
        tuple([dataclass_wizard.asdict(ctx) for ctx in contexts]),  # type: ignore
    )

dumps(collection)

Serializes a rats.app_context.Collection instance to a json string.

Parameters:

Name Type Description Default
collection Collection[Any]

collection of serializable services wanting to be shared between machines.

 required
Source code in rats/app_context/_collection.py 
152
153
154
155
156
157
158
159
def dumps(collection: Collection[Any]) -> str:
    """
    Serializes a [rats.app_context.Collection][] instance to a json string.

    Args:
        collection: collection of serializable services wanting to be shared between machines.
    """
    return json.dumps(dataclass_wizard.asdict(collection), indent=4)  # type: ignore

loads(context)

Transforms a json string into a rats.app_context.Collection instance.

This function does not attempt to validate the mapping between services and their values.

Parameters:

Name Type Description Default
context str

json string containing an items key of service_id, values mappings.

 required
Source code in rats/app_context/_collection.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def loads(context: str) -> Collection[Any]:
    """
    Transforms a json string into a [rats.app_context.Collection][] instance.

    This function does not attempt to validate the mapping between services and their values.

    Args:
        context: json string containing an `items` key of `service_id`, `values` mappings.
    """
    data = json.loads(context)
    return Collection[Any](
        items=tuple(
            [
                Context(
                    service_id=apps.ServiceId(*x["service_id"]),
                    values=tuple(x["values"]),
                )
                for x in data.get("items", [])
            ]
        ),
    )