hat.sbs

Simple binary serializer

This implementation of SBS encoder/decoder translates between SBS types and Python types according to following translation table::

+----------+------------------+
| SBS type | Python type      |
+==========+==================+
| None     | NoneType         |
+----------+------------------+
| Boolean  | bool             |
+----------+------------------+
| Integer  | int              |
+----------+------------------+
| Float    | float            |
+----------+------------------+
| String   | str              |
+----------+------------------+
| Bytes    | bytes            |
+----------+------------------+
| Array    | list[Data]       |
+----------+------------------+
| Record   | dict[str, Data]  |
+----------+------------------+
| Choice   | tuple[str, Data] |
+----------+------------------+

Example usage of SBS serializer::

import hat.sbs

repo = hat.sbs.Repository('''
    module Module

    Entry(K, V) = Record {
        key: K
        value: V
    }

    T = Array(Optional(Entry(String, Integer)))
''')
data = [
    ('none', None),
    ('value', {
        'key': 'abc',
        'value': 123
    })
]
encoded_data = repo.encode('Module.T', data)
decoded_data = repo.decode('Module.T', encoded_data)
assert data == decoded_data

 1"""Simple binary serializer
 2
 3This implementation of SBS encoder/decoder translates between SBS types and
 4Python types according to following translation table::
 5
 6    +----------+------------------+
 7    | SBS type | Python type      |
 8    +==========+==================+
 9    | None     | NoneType         |
10    +----------+------------------+
11    | Boolean  | bool             |
12    +----------+------------------+
13    | Integer  | int              |
14    +----------+------------------+
15    | Float    | float            |
16    +----------+------------------+
17    | String   | str              |
18    +----------+------------------+
19    | Bytes    | bytes            |
20    +----------+------------------+
21    | Array    | list[Data]       |
22    +----------+------------------+
23    | Record   | dict[str, Data]  |
24    +----------+------------------+
25    | Choice   | tuple[str, Data] |
26    +----------+------------------+
27
28Example usage of SBS serializer::
29
30    import hat.sbs
31
32    repo = hat.sbs.Repository('''
33        module Module
34
35        Entry(K, V) = Record {
36            key: K
37            value: V
38        }
39
40        T = Array(Optional(Entry(String, Integer)))
41    ''')
42    data = [
43        ('none', None),
44        ('value', {
45            'key': 'abc',
46            'value': 123
47        })
48    ]
49    encoded_data = repo.encode('Module.T', data)
50    decoded_data = repo.decode('Module.T', encoded_data)
51    assert data == decoded_data
52
53"""
54
55from hat.sbs.common import Ref, Data
56from hat.sbs.repository import Repository
57from hat.sbs.serializer import (Serializer,
58                                CSerializer,
59                                PySerializer,
60                                DefaultSerializer)
61
62
63__all__ = ['Ref',
64           'Data',
65           'Repository',
66           'Serializer',
67           'CSerializer',
68           'PySerializer',
69           'DefaultSerializer']
class Ref(typing.NamedTuple): 
7class Ref(typing.NamedTuple):
8    module: str | None
9    name: str

Ref(module, name)

Ref(module: str | None, name: str)

Create new instance of Ref(module, name)

module: str | None

Alias for field number 0

name: str

Alias for field number 1

Data = typing.Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, typing.List[ForwardRef('Data')], typing.Dict[str, ForwardRef('Data')], typing.Tuple[str, ForwardRef('Data')]]
class Repository: 
14class Repository:
15    """SBS schema repository.
16
17    Supported initialization arguments:
18        * string containing sbs schema
19        * file path to .sbs file
20        * path to direcory recursivly searched for .sbs files
21        * other repository
22
23    """
24
25    def __init__(self,
26                 *args: typing.Union['Repository', pathlib.Path, str]):
27        self._modules = list(_parse_args(args))
28        self._refs = evaluator.evaluate_modules(self._modules)
29
30    def encode(self,
31               name: str | common.Ref,
32               value: common.Data, *,
33               serializer: type[Serializer] = DefaultSerializer
34               ) -> util.Bytes:
35        """Encode value."""
36        ref = _parse_name(name) if isinstance(name, str) else name
37        return serializer.encode(self._refs, ref, value)
38
39    def decode(self,
40               name: str | common.Ref,
41               data: util.Bytes, *,
42               serializer: type[Serializer] = DefaultSerializer
43               ) -> common.Data:
44        """Decode data."""
45        ref = _parse_name(name) if isinstance(name, str) else name
46        return serializer.decode(self._refs, ref, data)
47
48    def to_json(self) -> json.Data:
49        """Export repository content as json serializable data.
50
51        Entire repository content is exported as json serializable data.
52        New repository can be created from the exported content by using
53        :meth:`Repository.from_json`.
54
55        """
56        return [parser.module_to_json(module) for module in self._modules]
57
58    @staticmethod
59    def from_json(data: pathlib.PurePath | json.Data) -> 'Repository':
60        """Create new repository from content exported as json serializable
61        data.
62
63        Creates a new repository from content of another repository that was
64        exported by using :meth:`Repository.to_json`.
65
66        """
67        if isinstance(data, pathlib.PurePath):
68            data = json.decode_file(data)
69
70        repo = Repository()
71        repo._modules = [parser.module_from_json(i) for i in data]
72        repo._refs = evaluator.evaluate_modules(repo._modules)
73        return repo

SBS schema repository.

Supported initialization arguments:
  • string containing sbs schema
  • file path to .sbs file
  • path to direcory recursivly searched for .sbs files
  • other repository
Repository(*args: Union[Repository, pathlib.Path, str]) 
25    def __init__(self,
26                 *args: typing.Union['Repository', pathlib.Path, str]):
27        self._modules = list(_parse_args(args))
28        self._refs = evaluator.evaluate_modules(self._modules)
def encode( self, name: str | Ref, value: Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]], *, serializer: type[Serializer] = <class 'CSerializer'>) -> bytes | bytearray | memoryview: 
30    def encode(self,
31               name: str | common.Ref,
32               value: common.Data, *,
33               serializer: type[Serializer] = DefaultSerializer
34               ) -> util.Bytes:
35        """Encode value."""
36        ref = _parse_name(name) if isinstance(name, str) else name
37        return serializer.encode(self._refs, ref, value)

Encode value.

def decode( self, name: str | Ref, data: bytes | bytearray | memoryview, *, serializer: type[Serializer] = <class 'CSerializer'>) -> Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]: 
39    def decode(self,
40               name: str | common.Ref,
41               data: util.Bytes, *,
42               serializer: type[Serializer] = DefaultSerializer
43               ) -> common.Data:
44        """Decode data."""
45        ref = _parse_name(name) if isinstance(name, str) else name
46        return serializer.decode(self._refs, ref, data)

Decode data.

def to_json( self) -> Union[NoneType, bool, int, float, str, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')]]: 
48    def to_json(self) -> json.Data:
49        """Export repository content as json serializable data.
50
51        Entire repository content is exported as json serializable data.
52        New repository can be created from the exported content by using
53        :meth:`Repository.from_json`.
54
55        """
56        return [parser.module_to_json(module) for module in self._modules]

Export repository content as json serializable data.

Entire repository content is exported as json serializable data. New repository can be created from the exported content by using Repository.from_json().

@staticmethod
def from_json( data: Union[pathlib.PurePath, NoneType, bool, int, float, str, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')]]) -> Repository: 
58    @staticmethod
59    def from_json(data: pathlib.PurePath | json.Data) -> 'Repository':
60        """Create new repository from content exported as json serializable
61        data.
62
63        Creates a new repository from content of another repository that was
64        exported by using :meth:`Repository.to_json`.
65
66        """
67        if isinstance(data, pathlib.PurePath):
68            data = json.decode_file(data)
69
70        repo = Repository()
71        repo._modules = [parser.module_from_json(i) for i in data]
72        repo._refs = evaluator.evaluate_modules(repo._modules)
73        return repo

Create new repository from content exported as json serializable data.

Creates a new repository from content of another repository that was exported by using Repository.to_json().

class Serializer(abc.ABC): 
11class Serializer(abc.ABC):
12
13    @staticmethod
14    @abc.abstractmethod
15    def encode(refs: dict[Ref, Type],
16               t: Type,
17               value: Data
18               ) -> util.Bytes:
19        """Encode value"""
20
21    @staticmethod
22    @abc.abstractmethod
23    def decode(refs: dict[Ref, Type],
24               t: Type,
25               data: util.Bytes
26               ) -> Data:
27        """Decode data"""

Helper class that provides a standard way to create an ABC using inheritance.

@staticmethod
@abc.abstractmethod
def encode( refs: dict[Ref, Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType], t: Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType, value: Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Dict[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Tuple[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]]]) -> bytes | bytearray | memoryview: 
13    @staticmethod
14    @abc.abstractmethod
15    def encode(refs: dict[Ref, Type],
16               t: Type,
17               value: Data
18               ) -> util.Bytes:
19        """Encode value"""

Encode value

@staticmethod
@abc.abstractmethod
def decode( refs: dict[Ref, Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType], t: Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType, data: bytes | bytearray | memoryview) -> Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Dict[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Tuple[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]]]: 
21    @staticmethod
22    @abc.abstractmethod
23    def decode(refs: dict[Ref, Type],
24               t: Type,
25               data: util.Bytes
26               ) -> Data:
27        """Decode data"""

Decode data

class CSerializer(hat.sbs.Serializer): 
11class CSerializer(common.Serializer):
12    """Serializer implementation in C"""
13
14    def encode(refs, t, value):
15        if not _cserializer:
16            raise Exception('implementation not available')
17
18        return _cserializer.encode(refs, t, value)
19
20    def decode(refs, t, data):
21        if not _cserializer:
22            raise Exception('implementation not available')
23
24        return _cserializer.decode(refs, t, memoryview(data))

Serializer implementation in C

def encode(refs, t, value): 
14    def encode(refs, t, value):
15        if not _cserializer:
16            raise Exception('implementation not available')
17
18        return _cserializer.encode(refs, t, value)

Encode value

def decode(refs, t, data): 
20    def decode(refs, t, data):
21        if not _cserializer:
22            raise Exception('implementation not available')
23
24        return _cserializer.decode(refs, t, memoryview(data))

Decode data

class PySerializer(hat.sbs.Serializer): 
 8class PySerializer(common.Serializer):
 9    """Serializer implementation in Python"""
10
11    def encode(refs, t, value):
12        return bytes(_encode_generic(refs, t, value))
13
14    def decode(refs, t, data):
15        value, _ = _decode_generic(refs, t, memoryview(data))
16        return value

Serializer implementation in Python

def encode(refs, t, value): 
11    def encode(refs, t, value):
12        return bytes(_encode_generic(refs, t, value))

Encode value

def decode(refs, t, data): 
14    def decode(refs, t, data):
15        value, _ = _decode_generic(refs, t, memoryview(data))
16        return value

Decode data

DefaultSerializer = <class 'CSerializer'>