Context Dict#

This module defines classes for lazy context data loading.

_get_hash(string)[source]#

Return type:: bytes

class ContextDict(**data)[source]#

Bases: ABC, BaseModel

Dictionary-like structure for storing dialog data spanning multiple turns in a context storage. It holds all the possible keys, but may not store all the values locally. Values not stored locally will be loaded upon querying.

Keys of the dictionary are turn ids. Get, set and delete operations support slices in addition to regular single-key operation. Examples:

await ctx.labels[0] returns label at turn 0 (start label).
await ctx.requests[1:4] returns the first 3 requests (requests at turns 1, 2, 3).
ctx.responses[2:5:2] = Message("2"), Message("4") sets responses at turns 2 and 4.

Get operation is asynchronous (and requires await) since it calls _load_items() in order to get items that need to be loaded from DB.

_items: Dict[int, BaseModel] = ModelPrivateAttr(default=PydanticUndefined, default_factory=<class 'dict'>)#: Dictionary of already loaded from storage items.

_hashes: Dict[int, int] = ModelPrivateAttr(default=PydanticUndefined, default_factory=<class 'dict'>)#: Hashes of the loaded items (as they were upon loading), only used if rewrite_existing flag is enabled.

_keys: Set[int] = ModelPrivateAttr(default=PydanticUndefined, default_factory=<class 'set'>)#: All the item keys available either in storage or locally.

_added: Set[int] = ModelPrivateAttr(default=PydanticUndefined, default_factory=<class 'set'>)#: Keys added locally (need to be synchronized with the storage). Synchronization happens whenever store is called (which is done at the end of every turn).

_removed: Set[int] = ModelPrivateAttr(default=PydanticUndefined, default_factory=<class 'set'>)#: Keys removed locally (need to be synchronized with the storage). Synchronization happens whenever store is called (which is done at the end of every turn).

_storage: Optional[DBContextStorage] = ModelPrivateAttr()#: Context storage for item synchronization.

_ctx_id: str = ModelPrivateAttr(default=PydanticUndefined, default_factory=<class 'str'>)#: Corresponding context ID.

_field_name: str = ModelPrivateAttr(default=PydanticUndefined, default_factory=<class 'str'>)#: Name of the field in the context storage that is represented by the given dict.

abstract property _value_type: TypeAdapter[BaseModel]#

async classmethod new(storage, id, field)[source]#

Create a new context dict, without connecting it to the context storage. No keys or items will be loaded, but any newly added items will be available for synchronization. Should be used when we are sure that context with given ID does not exist in the storage.

Parameters:

storage (DBContextStorage) – Context storage, where the new items will be added.
id (str) – Newly created context ID.
field (str) – Current dict field name.

Return type:

ContextDict

Returns:

New “disconnected” context dict.

async classmethod connected(storage, id, field)[source]#

Create a new context dict, connecting it to the context storage. All the keys and some items will be loaded, all the other items will be available for synchronization. Also hashes will be calculated for the initially loaded items for modification tracking.

Parameters:

storage (DBContextStorage) – Context storage, keeping the current context.
id (str) – Newly created context ID.
field (str) – Current dict field name.

Return type:

ContextDict

Returns:

New “connected” context dict.

async _load_items(keys)[source]#

Load items for the given keys from the connected context storage. Update the _items and _hashes fields if necessary. NB! If not all the requested items are available, only the successfully loaded will be updated and no error will be raised.

Parameters:: keys (List[int]) – The requested key array.
Return type:: None

async get(key, default=None)[source]#

Get one or many items from the dict. Asynchronously load missing ones, if context storage is connected.

Parameters:

key – Key or an iterable of keys for item retrieving.
default – Default value. Default is returned when key is a single key that is not in the dict. If key is an iterable of keys, default is returned instead of all the values that are not in the dict.

Returns:

A single value if key is a single key. Tuple of values if key is an iterable.

keys()[source]#

Return type:: List[int]

async values()[source]#

Return type:: List[BaseModel]

async items()[source]#

Return type:: List[Tuple[int, BaseModel]]

async pop(key, default=None)[source]#

Return type:: BaseModel

clear()[source]#

Return type:: None

async update(other=(), /, **kwds)[source]#

Return type:: None

async setdefault(key, default=None)[source]#

Return type:: BaseModel

_validate_model(handler, _)[source]#

Return type:: ContextDict

_serialize_model_base(to_bytes=False)[source]#

Return type:: Dict[int, Union[BaseModel, bytes]]

_serialize_model()[source]#

Return type:: Dict[int, BaseModel]

extract_sync()[source]#

Synchronize dict state with the connected storage, extract the data that should be updated. Update added and removed elements, also update modified ones if rewrite_existing flag is enabled. Raise an error if no storage is connected.

Return type:: Tuple[str, List[Tuple[int, bytes]], List[int]]

model_post_init(context, /)#

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Return type:: None

Args:: self: The BaseModel instance. context: The context.

class LabelContextDict(**data)[source]#

Bases: ContextDict

Context dictionary for storing AbsoluteNodeLabel types.

_items: Dict[int, AbsoluteNodeLabel] = ModelPrivateAttr(default=PydanticUndefined)#: Dictionary of already loaded from storage items.

property _value_type: TypeAdapter[AbsoluteNodeLabel]#

model_post_init(context: Any, /) → None#

We need to both initialize private attributes and call the user-defined model_post_init method.

Return type:: None

async get(key, default=None)[source]#

Get one or many items from the dict. Asynchronously load missing ones, if context storage is connected.

Parameters:

key – Key or an iterable of keys for item retrieving.
default – Default value. Default is returned when key is a single key that is not in the dict. If key is an iterable of keys, default is returned instead of all the values that are not in the dict.

Returns:

A single value if key is a single key. Tuple of values if key is an iterable.

async values()[source]#

Return type:: List[AbsoluteNodeLabel]

async items()[source]#

Return type:: List[Tuple[int, AbsoluteNodeLabel]]

async pop(key, default=None)[source]#

Return type:: AbsoluteNodeLabel

async setdefault(key, default=None)[source]#

Return type:: AbsoluteNodeLabel

class MessageContextDict(**data)[source]#

Bases: ContextDict

Context dictionary for storing Message types.

model_post_init(context: Any, /) → None#

We need to both initialize private attributes and call the user-defined model_post_init method.

Return type:: None

_items: Dict[int, Message] = ModelPrivateAttr(default=PydanticUndefined)#: Dictionary of already loaded from storage items.

property _value_type: TypeAdapter[Message]#

async get(key, default=None)[source]#

Get one or many items from the dict. Asynchronously load missing ones, if context storage is connected.

Parameters:

key – Key or an iterable of keys for item retrieving.
default – Default value. Default is returned when key is a single key that is not in the dict. If key is an iterable of keys, default is returned instead of all the values that are not in the dict.

Returns:

A single value if key is a single key. Tuple of values if key is an iterable.

async values()[source]#

Return type:: List[Message]

async items()[source]#

Return type:: List[Tuple[int, Message]]

async pop(key, default=None)[source]#

Return type:: Message

async setdefault(key, default=None)[source]#

Return type:: Message