cardano streamed transactions

lazy_cbor_collection.py

class HashBuilderCollection:
    def __init__(self, size: int) -> None:
        self.size = size
        self.remaining = size
        self.hash_fn: HashContext | None = None
        self.parent: "HashBuilderCollection" | None = None
        self.have_nested_child = False

    def start(self, hash_fn: HashContext) -> "HashBuilderCollection":
        self.hash_fn = hash_fn
        return self

    def _insert_into_parent(self, parent: "HashBuilderCollection") -> None:
        self.parent = parent
        assert parent.hash_fn is not None
        self.hash_fn = parent.hash_fn
        parent.have_nested_child = True
        self.hash_fn.update(self._header_bytes())

    def _do_enter_item(self) -> None:
        assert self.hash_fn is not None
        assert self.remaining > 0
        if self.have_nested_child:
            raise RuntimeError  # can't add item until child is done

        self.remaing -= 1

    def _hash_item(self, item: Any) -> None:
        assert self.hash_fn is not None
        for chunk in cbor.encode_streamed(item):
            self.hash_fn.update(chunk)

    def _header_bytes(self) -> bytes:
        raise NotImplementedError

    def finish(self) -> None:
        if self.remaining != 0:
            raise RuntimeError  # not all items were added
        if self.parent is not None:
            self.parent.have_nested_child = False
        self.hash_fn = None
        self.parent = None

    def __enter__(self) -> "HashBuilderCollection":
        assert self.hash_fn is not None
        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        self.finish()


class LazyList(HashBuilderCollection, Generic[T]):
    def append(self, item: T) -> T:
        self._do_enter_item()
        if isinstance(item, HashBuilderCollection):
            item._insert_into_parent(self)
        else:
            self._hash_item(item)

        return item

    def _header_bytes(self) -> bytes:
        return cbor.create_array_header(self.size)


class LazyDict(HashBuilderCollection, Generic[K, V]):
    def add(self, key: K, value: V) -> V:
        self._do_enter_item()
        # enter key, this must not nest
        assert not isinstance(key, HashBuilderCollection)
        self._hash_item(key)
        # enter value, this can nest
        if isinstance(value, HashBuilderCollection):
            value._insert_into_parent(self)
        else:
            self._hash_item(value)

        return value

    def _header_bytes(self) -> bytes:
        return cbor.create_map_header(self.size)