API#

local_migrator.REGISTER = <local_migrator._class_register.MigrationRegistration object>#: Implementation of class register to storage information needed for migration from previous version.

local_migrator.MigrationInfo#: alias of Tuple[Version, Callable[[Dict[str, Any]], Dict[str, Any]]]

local_migrator.object_encoder(obj)[source]#

Function changing supported types to basic python types supported by most serializers and which could be restored by nme_object_hook() function.

Supported types are:

enum.Enum
dataclasses.dataclass()
numpy.ndarray
pydantic.BaseModel
numpy.integer (change to pure int)
numpy.floating (change to pure float)
pathlib.Path (Serialized to string)
Any class with an as_dict method. This method should return a dictionary of valid constructor arguments.

Parameters:: obj (Any) – object to be encoded.
Returns:: encoded object for supported types. Otherwise None.

class local_migrator.Encoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]#

Bases: JSONEncoder

JSONEncoder subclass for serializing Python objects into JSON. For list of supported types check nme_object_encoder() function.

default(o)[source]#: Implementation that calls nme_object_encoder() function.

class local_migrator.MigrationRegistration[source]#

Bases: object

Implementation of class register to storage information needed for migration from previous version.

allow_errors_in_values(cls)[source]#

Check if class should allow errors in values.

Parameters:: cls (Union[str, Type]) – class or full qualified path to class
Return type:: bool

get_class(class_str)[source]#

Get class base of qualified name. Could be done using current or old path.

Qualified name is determined using class_to_str() and old path comes from the old_paths argument of register() method.

Return type:: Type

get_version(cls)[source]#

For a given class return version with which given class was registered using register()

Return type:: Version

migrate_data(cls, class_str_to_version_dkt, data)[source]#

Apply migrations base on register state. Current implementation does not support multiple inheritance.

Parameters:

cls (Union[str, Type]) – fully qualified class path
class_str_to_version_dkt (Dict[str, Union[str, Version]]) – for each parent class information about version during serialization. If class is absent from this dict then assumed version is “0.0.0”
data (Dict[str, Any]) – dict of kwargs to constructor of class

Return type:

Dict[str, Any]

register(cls=None, version='0.0.0', migrations=None, old_paths=None, use_parent_migrations=True, allow_errors_in_values=False)[source]#

Parameters:

cls (Optional[Type]) – class to be registered
version (Union[str, Version]) – current version of class
migrations (Optional[List[Tuple[Union[str, Version], Callable[[Dict[str, Any]], Dict[str, Any]]]]]) – list of migrations for deserialize old version
old_paths (Optional[List[str]]) – old name of class with modules
use_parent_migrations (bool) – if migrations from parent class should be applied when deserialized object
allow_errors_in_values (bool) – if errors in constructor kwargs should be ignored. Added to not block creating of custom Mapping class that could contain broken items.

Return type:

Union[Type[TypeVar(T)], Callable[[Type[TypeVar(T)]], Type[TypeVar(T)]]]

Returns:

class itself if cls parameter is provided. Otherwise, one argument function which will consume Type to be registered.

use_parent_migrations(name)[source]#

Check if parent migrations should be used.

Parameters:: name (str) – full qualified path to class
Return type:: bool
Returns:: information if parent class migrations should be applied.

local_migrator.NMEEncoder#: alias of Encoder

local_migrator.cbor_decoder(*args)[source]#

Cbor decoder hook. Use nme_object_hook() to decode objects.

Callback signature changed across cbor2 releases:

Older versions call it as (decoder, value).
Newer versions call it as (value, immutable).

This wrapper supports both conventions.

Parameters:: args – callback arguments from cbor2

Examples:

with open(path_to_file, "rb") as f_p:
    data = cbor2.load(f_p, object_hook=nme_cbor_decoder)

local_migrator.cbor_encoder(encoder, value)[source]#

Cbor encoder hook. Use nme_object_encoder() to encode objects.

Parameters:

encoder – cbor2.Encoder
value – object to be encoded

Examples:

with open(path_to_file, "wb") as f_p:
    cbor2.dump(data, f_p, default=nme_cbor_encoder)

local_migrator.check_for_errors_in_dkt_values(dkt)[source]#

Function checking if any of values in dict contains "__error__" key.

Parameters:: dkt (dict) – dictionary to check.
Return type:: List[str]
Returns:: list of keys that value is dict containing "__error__" key.

local_migrator.class_to_str(cls)[source]#

Get full qualified name for a given class.

Return type:: str

local_migrator.nme_cbor_decoder(*args)#

Cbor decoder hook. Use nme_object_hook() to decode objects.

Callback signature changed across cbor2 releases:

Older versions call it as (decoder, value).
Newer versions call it as (value, immutable).

This wrapper supports both conventions.

Parameters:: args – callback arguments from cbor2

Examples:

with open(path_to_file, "rb") as f_p:
    data = cbor2.load(f_p, object_hook=nme_cbor_decoder)

local_migrator.nme_cbor_encoder(encoder, value)#

Cbor encoder hook. Use nme_object_encoder() to encode objects.

Parameters:

encoder – cbor2.Encoder
value – object to be encoded

Examples:

with open(path_to_file, "wb") as f_p:
    cbor2.dump(data, f_p, default=nme_cbor_encoder)

local_migrator.nme_object_hook(dkt)#

Function restoring supported types from nme_object_encoder() function output.

If dkt does not contain __class__ key, it is returned as is.

If the restoring object fails then function return dict with "__error__" key.

Parameters:: dkt (dict) – dictionary with data to restore.
Return type:: Any

local_migrator.object_hook(dkt)[source]#

Function restoring supported types from nme_object_encoder() function output.

If dkt does not contain __class__ key, it is returned as is.

If the restoring object fails then function return dict with "__error__" key.

Parameters:: dkt (dict) – dictionary with data to restore.
Return type:: Any

local_migrator.register_class(cls=None, version='0.0.0', migrations=None, old_paths=None, use_parent_migrations=True, allow_errors_in_values=False)[source]#

This is wrapper for call MigrationRegistration.register() of default register instance. Please see its documentation for details.

Parameters:

cls (Optional[Type[TypeVar(T)]]) – class to be registered
version (Union[str, Version]) – current version of class
migrations (Optional[List[Tuple[Union[str, Version], Callable[[Dict[str, Any]], Dict[str, Any]]]]]) – list of migrations for deserialize old version
old_paths (Optional[List[str]]) – old name of class with modules
use_parent_migrations (bool) – if migrations from parent class should be applied when deserialized object
allow_errors_in_values (bool) – if errors in constructor kwargs should be ignored. Added to not block creating of custom Mapping class that could contain broken items.

Return type:

Union[Type[TypeVar(T)], Callable[[Type[TypeVar(T)]], Type[TypeVar(T)]]]

Returns:

class itself if cls parameter is provided. Otherwise, one argument function which will consume Type to be registered.

Examples:

@register_class(version="0.0.1", migrations=[("0.0.1", rename_key("value", "value1"))])
class DataClass:
    def __init__(value1, value2)
        self.value1 = value1
        self.value2 = value2

or:

class DataClass2:
    def __init__(value1, value2)
        self.value1 = value1
        self.value2 = value2

register_class(DataClass2, version="0.0.1", migrations=[("0.0.1", rename_key("value", "value1"))])

local_migrator.rename_key(from_key, to_key, optional=False)[source]#

simple migration function for rename fields

Parameters:

from_key (str) – original name
to_key (str) – destination name
optional – if migration is required (for backward compatibility)

Return type:

Callable[[Dict[str, Any]], Dict[str, Any]]

Returns:

migration function

local_migrator.update_argument(argument_name)[source]#

This is decorator for move conversion of dict to class outside function code. It first inspects function signature to determine type th which argument should be converted. Then, if argument is passed as dict then all migrations from REGISTER all applied, then object is constructed and replace base one.

Parameters:: argument_name – name of argument which should be converted

Example:

@register_class(version="0.0.1", migrations=[("0.0.1", rename_key("value", "value1"))])
class DataClass:
    def __init__(value1, value2)
        self.value1 = value1
        self.value2 = value2

@update_argument("arg")
def some_func(arg: DataClass):
    print(arg.value1)

some_func({"value": 1, "value2": 5})