File#

The File module provides simple file-based versions of the DBContextStorage class such as json, pickle and shelve.

class SerializableStorage(**data)[source]#

Bases: BaseModel

A special serializable database implementation. One element of this class will be used to store all the contexts, read and written to file on every turn.

main: Dict[str, ContextMainInfo]#
turns: List[Tuple[str, str, int, Optional[bytes]]]#
class FileContextStorage(path='', rewrite_existing=False, partial_read_config=None)[source]#

Bases: DBContextStorage, ABC

Implements DBContextStorage with any file-based storage format.

Parameters:

  • path (str) – Target file URI.

  • rewrite_existing (bool) – Whether TURNS modified locally should be updated in database or not.

  • partial_read_config (Optional[Dict[Literal['labels', 'requests', 'responses'], Union[Literal['__all__'], int, Set[int]]]]) – Dictionary of subscripts for all possible turn items.

is_concurrent: bool = False#
abstract async _save(data)[source]#
Return type:

None

abstract async _load()[source]#
Return type:

SerializableStorage

async _connect()[source]#
async _load_main_info(ctx_id)[source]#
Return type:

Optional[ContextMainInfo]

async _update_context(ctx_id, ctx_info, field_info)[source]#
Return type:

None

async _delete_context(ctx_id)[source]#
Return type:

None

async _load_field_latest(ctx_id, field_name)[source]#
Return type:

List[Tuple[int, bytes]]

async _load_field_keys(ctx_id, field_name)[source]#
Return type:

List[int]

async _load_field_items(ctx_id, field_name, keys)[source]#
Return type:

List[Tuple[int, bytes]]

async _clear_all()[source]#
Return type:

None

class JSONContextStorage(path='', rewrite_existing=False, partial_read_config=None)[source]#

Bases: FileContextStorage

Implements DBContextStorage with json as the storage format.

Parameters:

  • path (str) – Target file URI. Example: json://file.json.

  • rewrite_existing (bool) – Whether TURNS modified locally should be updated in database or not.

  • partial_read_config (Optional[Dict[Literal['labels', 'requests', 'responses'], Union[Literal['__all__'], int, Set[int]]]]) – Dictionary of subscripts for all possible turn items.

async _save(data)[source]#
Return type:

None

async _load()[source]#
Return type:

SerializableStorage

class PickleContextStorage(path='', rewrite_existing=False, partial_read_config=None)[source]#

Bases: FileContextStorage

Implements DBContextStorage with pickle as the storage format.

Parameters:

  • path (str) – Target file URI. Example: pickle://file.pkl.

  • rewrite_existing (bool) – Whether TURNS modified locally should be updated in database or not.

  • partial_read_config (Optional[Dict[Literal['labels', 'requests', 'responses'], Union[Literal['__all__'], int, Set[int]]]]) – Dictionary of subscripts for all possible turn items.

async _save(data)[source]#
Return type:

None

async _load()[source]#
Return type:

SerializableStorage

class ShelveContextStorage(path='', rewrite_existing=False, partial_read_config=None)[source]#

Bases: FileContextStorage

Implements DBContextStorage with shelve as the storage format.

Parameters:

  • path (str) – Target file URI. Example: shelve://file.shlv.

  • rewrite_existing (bool) – Whether TURNS modified locally should be updated in database or not.

  • partial_read_config (Optional[Dict[Literal['labels', 'requests', 'responses'], Union[Literal['__all__'], int, Set[int]]]]) – Dictionary of subscripts for all possible turn items.

_SHELVE_ROOT = 'root'#
async _save(data)[source]#
Return type:

None

async _load()[source]#
Return type:

SerializableStorage