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

Serializer:
- decode
- encode

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

Serializer:
- decode
- encode

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"""