Module 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 |
+==========+==================+
| Boolean | bool |
+----------+------------------+
| Integer | int |
+----------+------------------+
| Float | float |
+----------+------------------+
| String | str |
+----------+------------------+
| Bytes | bytes |
+----------+------------------+
| Array | List[Data] |
+----------+------------------+
| Tuple | Dict[str, Data] |
+----------+------------------+
| Union | Tuple[str, Data] |
+----------+------------------+
SBS Tuple and Union types without elements are translated to None
.
Example usage of SBS serializer::
import hat.sbs
repo = hat.sbs.Repository('''
module Module
Entry(K, V) = Tuple {
key: K
value: V
}
T = Array(Maybe(Entry(String, Integer)))
''')
data = [
('Nothing', None),
('Just', {
'key': 'abc',
'value': 123
})
]
encoded_data = repo.encode('Module', 'T', data)
decoded_data = repo.decode('Module', 'T', encoded_data)
assert data == decoded_data
Expand source code
"""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 |
+==========+==================+
| Boolean | bool |
+----------+------------------+
| Integer | int |
+----------+------------------+
| Float | float |
+----------+------------------+
| String | str |
+----------+------------------+
| Bytes | bytes |
+----------+------------------+
| Array | List[Data] |
+----------+------------------+
| Tuple | Dict[str, Data] |
+----------+------------------+
| Union | Tuple[str, Data] |
+----------+------------------+
SBS Tuple and Union types without elements are translated to ``None``.
Example usage of SBS serializer::
import hat.sbs
repo = hat.sbs.Repository('''
module Module
Entry(K, V) = Tuple {
key: K
value: V
}
T = Array(Maybe(Entry(String, Integer)))
''')
data = [
('Nothing', None),
('Just', {
'key': 'abc',
'value': 123
})
]
encoded_data = repo.encode('Module', 'T', data)
decoded_data = repo.decode('Module', 'T', encoded_data)
assert data == decoded_data
"""
from hat.sbs.common import Data
from hat.sbs.repository import Repository
from hat.sbs.serializer import (Serializer,
CSerializer,
PySerializer)
__all__ = ['Repository', 'Data', 'Serializer', 'CSerializer', 'PySerializer']
Sub-modules
hat.sbs.common
hat.sbs.evaluator
hat.sbs.parser
hat.sbs.repository
hat.sbs.serializer
Classes
class CSerializer
-
Serializer implementation in C
Expand source code
class CSerializer(Serializer): """Serializer implementation in C""" def encode(refs, t, value): if not _cserializer: raise Exception('implementation not available') return _cserializer.encode(refs, t, value) def decode(refs, t, data): if not _cserializer: raise Exception('implementation not available') return _cserializer.decode(refs, t, data)
Ancestors
- Serializer
- abc.ABC
Inherited members
class PySerializer
-
Serializer implementation in Python
Expand source code
class PySerializer(Serializer): """Serializer implementation in Python""" def encode(refs, t, value): return _pyserializer.encode(refs, t, value) def decode(refs, t, data): return _pyserializer.decode(refs, t, data)
Ancestors
- Serializer
- abc.ABC
Inherited members
class Repository (*args: typing.Union[ForwardRef('Repository'), pathlib.Path, str], serializer=hat.sbs.serializer.CSerializer)
-
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
Expand source code
class Repository: """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 """ def __init__(self, *args: typing.Union['Repository', pathlib.Path, str], serializer=serializer.CSerializer): self._serializer = serializer self._modules = list(_parse_args(args)) self._refs = evaluator.evaluate_modules(self._modules) def encode(self, module_name: typing.Optional[str], type_name: str, value: common.Data ) -> bytes: """Encode value.""" ref = common.Ref(module_name, type_name) return self._serializer.encode(self._refs, ref, value) def decode(self, module_name: typing.Optional[str], type_name: str, data: typing.Union[bytes, bytearray, memoryview] ) -> common.Data: """Decode data.""" ref = common.Ref(module_name, type_name) return self._serializer.decode(self._refs, ref, memoryview(data)) def to_json(self) -> json.Data: """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 :meth:`Repository.from_json`. """ return [parser.module_to_json(module) for module in self._modules] @staticmethod def from_json(data: typing.Union[pathlib.PurePath, common.Data], *, serializer=serializer.CSerializer ) -> 'Repository': """Create new repository from content exported as json serializable data. Creates a new repository from content of another repository that was exported by using :meth:`Repository.to_json`. """ if isinstance(data, pathlib.PurePath): data = json.decode_file(data) repo = Repository(serializer=serializer) repo._modules = [parser.module_from_json(i) for i in data] repo._refs = evaluator.evaluate_modules(repo._modules) return repo
Static methods
def from_json(data: typing.Union[pathlib.PurePath, bool, int, float, str, bytes, typing.List[ForwardRef('Data')], typing.Dict[str, ForwardRef('Data')], typing.Tuple[str, ForwardRef('Data')]], *, serializer=hat.sbs.serializer.CSerializer) ‑> Repository
-
Create new repository from content exported as json serializable data.
Creates a new repository from content of another repository that was exported by using :meth:
Repository.to_json()
.Expand source code
@staticmethod def from_json(data: typing.Union[pathlib.PurePath, common.Data], *, serializer=serializer.CSerializer ) -> 'Repository': """Create new repository from content exported as json serializable data. Creates a new repository from content of another repository that was exported by using :meth:`Repository.to_json`. """ if isinstance(data, pathlib.PurePath): data = json.decode_file(data) repo = Repository(serializer=serializer) repo._modules = [parser.module_from_json(i) for i in data] repo._refs = evaluator.evaluate_modules(repo._modules) return repo
Methods
def decode(self, module_name: typing.Union[str, NoneType], type_name: str, data: typing.Union[bytes, bytearray, memoryview]) ‑> typing.Union[bool, int, float, str, bytes, typing.List[Data], typing.Dict[str, Data], typing.Tuple[str, Data]]
-
Decode data.
Expand source code
def decode(self, module_name: typing.Optional[str], type_name: str, data: typing.Union[bytes, bytearray, memoryview] ) -> common.Data: """Decode data.""" ref = common.Ref(module_name, type_name) return self._serializer.decode(self._refs, ref, memoryview(data))
def encode(self, module_name: typing.Union[str, NoneType], type_name: str, value: typing.Union[bool, int, float, str, bytes, typing.List[ForwardRef('Data')], typing.Dict[str, ForwardRef('Data')], typing.Tuple[str, ForwardRef('Data')]]) ‑> bytes
-
Encode value.
Expand source code
def encode(self, module_name: typing.Optional[str], type_name: str, value: common.Data ) -> bytes: """Encode value.""" ref = common.Ref(module_name, type_name) return self._serializer.encode(self._refs, ref, value)
def to_json(self) ‑> typing.Union[NoneType, bool, int, float, str, typing.List[Data], typing.Dict[str, Data]]
-
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 :meth:
Repository.from_json()
.Expand source code
def to_json(self) -> json.Data: """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 :meth:`Repository.from_json`. """ return [parser.module_to_json(module) for module in self._modules]
class Serializer
-
Helper class that provides a standard way to create an ABC using inheritance.
Expand source code
class Serializer(abc.ABC): @staticmethod @abc.abstractmethod def encode(refs: typing.Dict[common.Ref, common.Type], t: common.Type, value: common.Data ) -> bytes: """Encode value""" @staticmethod @abc.abstractmethod def decode(refs: typing.Dict[common.Ref, common.Type], t: common.Type, data: memoryview ) -> common.Data: """Decode data"""
Ancestors
- abc.ABC
Subclasses
Static methods
def decode(refs: typing.Dict[Ref, typing.Union[Ref, BooleanType, IntegerType, FloatType, StringType, BytesType, ArrayType, TupleType, UnionType]], t: typing.Union[Ref, BooleanType, IntegerType, FloatType, StringType, BytesType, ArrayType, TupleType, UnionType], data: memoryview) ‑> typing.Union[bool, int, float, str, bytes, typing.List[Data], typing.Dict[str, Data], typing.Tuple[str, Data]]
-
Decode data
Expand source code
@staticmethod @abc.abstractmethod def decode(refs: typing.Dict[common.Ref, common.Type], t: common.Type, data: memoryview ) -> common.Data: """Decode data"""
def encode(refs: typing.Dict[Ref, typing.Union[Ref, BooleanType, IntegerType, FloatType, StringType, BytesType, ArrayType, TupleType, UnionType]], t: typing.Union[Ref, BooleanType, IntegerType, FloatType, StringType, BytesType, ArrayType, TupleType, UnionType], value: typing.Union[bool, int, float, str, bytes, typing.List[ForwardRef('Data')], typing.Dict[str, ForwardRef('Data')], typing.Tuple[str, ForwardRef('Data')]]) ‑> bytes
-
Encode value
Expand source code
@staticmethod @abc.abstractmethod def encode(refs: typing.Dict[common.Ref, common.Type], t: common.Type, value: common.Data ) -> bytes: """Encode value"""