Metadata-Version: 2.4
Name: sansio_tools
Version: 1.2.0
Summary: Tools for building Sans IO libraries
Author-email: Eduard Christian Dumitrescu <eduard.c.dumitrescu@gmail.com>
Maintainer-email: Eduard Christian Dumitrescu <eduard.c.dumitrescu@gmail.com>
License: MIT
Project-URL: documentation, https://www.ecd.space/doc/sansio_tools/latest/
Project-URL: source, https://hydra.ecd.space/ecd/sansio-tools/
Project-URL: changelog, https://hydra.ecd.space/ecd/sansio-tools/file?name=CHANGELOG.md&ci=trunk
Classifier: Programming Language :: Python :: 3
Description-Content-Type: text/markdown
License-File: LICENSE
License-File: AUTHORS.md
Dynamic: license-file

# sansio_tools

Useful tools for building [Sans IO](https://sans-io.readthedocs.io/)-based libraries.

## {class}`~.queue.BytesQueue`

Example code:

```python
from sansio_tools.queue import BytesQueue

q = BytesQueue()

q.append(b"012345")

print(bytes(q.peekleft(2)))  # b"01"

print(bytes(q.popleft(3)))  # b"012"

# put the bytes back into the queue!
q.appendleft(b"012")

print(bytes(q.popleft(4)))  # b"0123"
```

## {class}`~.parser.BinaryParser`

Example code:

```python
from struct import Struct
from sansio_tools.parser import BinaryParser


# int8, uint32, uint64
my_struct = Struct("!bIQ")

def parser_generator(p: BinaryParser):
    b = yield from p.read_bytes(6)
    results.append(b)
    b = yield from p.read_bytes(6)
    results.append(b)

    lst = yield from p.read_struct(my_struct)
    results.append(lst)

    n = yield from p.read_int(2, "little", False)
    results.append(n)

    n = yield from p.read_variable_length_int_7bit(
        maximum_length=10,
        byteorder="big",
        continuation_bit_value=True,
        require_canonical=True,
    )
    results.append(n)
    return "ok!"

results = []
p = BinaryParser(parser_generator)
p.feed(b"hell")
p.feed(b"o world!\xff")

# receive partial results as they become available!
assert results == [b"hello ", b"world!"]
assert not p.generator_exited
results.clear()

p.feed(b"\xff\x00\x00\x00\xff\x00\x00\x00\x00\x00\x00\x00\x12\x34")
p.feed(b"\x81\x80\x80")

assert results == [(-1, 0xff << 24, 0xff << 56), 0x3412]
assert not p.generator_exited
results.clear()

p.feed(b"\x80\x05")
p.feed(None)  # signal eof

assert results == [(1 << (7 * 4)) + 5]
assert p.generator_exited
assert p.generator_return_value == "ok!"
```

## {class}`~.parser.StringParser`

{class}`.StringParser` is the text counterpart of {class}`.BinaryParser`: a generator-based parser for streams of
`str` chunks (e.g. data received from a socket, decoded to text). Its main method,
{meth}`~.StringParser.read_character_ranges()`, reads a string made up of allowed characters, which you declare
with a {class}`.CharacterRanges` object. Reading stops at end of stream, when the maximum length is reached, or at
the first disallowed character, and the {class}`.StringStopReason` returned alongside the string tells you
which of those happened. Any terminating character is left in the queue so your parser can consume
or inspect it next.

Example code:

```python
from sansio_tools.parser import CharacterRanges, StringParser, StringStopReason

# Declare which characters are allowed in a "word" (here: a-z and 0-9).
word_chars = CharacterRanges(((97, 123), (48, 58)))

results = []

def parser_generator(p: StringParser):
    # Read words, one at a time, separated by any non-word character.
    while True:
        # Stop once the stream has ended and the buffer is empty.
        if p.eof and not p.queue:
            break

        # Read 1 to 10 word characters. The terminating character (if any)
        # is left in the queue, and the stop reason tells us why reading stopped.
        text, reason = yield from p.read_character_ranges(word_chars, min_length=1, max_length=10)
        results.append((text, reason))

        if reason is StringStopReason.FORBIDDEN:
            # A non-word character ended the word; consume it as a separator.
            p.queue.popleft(1)

p = StringParser(parser_generator)

# Feed the input in chunks; words are assembled across feed boundaries.
p.feed("he")
p.feed("llo,")
assert results == [("hello", StringStopReason.FORBIDDEN)]

p.feed("wor")
p.feed("ld!")
assert results == [
    ("hello", StringStopReason.FORBIDDEN),
    ("world", StringStopReason.FORBIDDEN),
]

p.feed("abcdefg")
p.feed("hijklmnop")
assert results[-1] == ("abcdefghij", StringStopReason.MAX_LENGTH)

p.feed(None)  # signal eof
assert results == [
    ("hello", StringStopReason.FORBIDDEN),
    ("world", StringStopReason.FORBIDDEN),
    ("abcdefghij", StringStopReason.MAX_LENGTH),
    ("klmnop", StringStopReason.EOF),
]
assert p.generator_exited
```

Notes:

- If fewer than `min_length` matching characters are found before the string ends, {meth}`~.StringParser.read_character_ranges()`
  raises {class}`.UnexpectedCharacterError`.
- {meth}`~.StringParser.read_character_ranges_iter()` is the streaming variant: it yields each chunk of matching characters as it
  becomes available (yielding `None` whenever it needs more data) and returns a `StringStopReason` at the end.
  Use it when you want to process a long string incrementally instead of receiving it as one joined string.

## {class}`~.multihash.HashFunctionRegistry`

The `multihash` module implements the [multihash](https://github.com/multiformats/multihash) and
[multibase](https://github.com/multiformats/multibase) formats. A *multihash* is a self-describing hash:
it embeds the hash function code and the digest size as varint prefixes in front of the raw digest, so
the value can be interpreted without any out-of-band information about which algorithm produced it.

A ready-to-use {data}`~.multihash.registry` instance ships with a set of common hash functions (SHA-2,
SHA-3, SHAKE-256, and BLAKE2). Use {attr}`~.multihash.HashFunctionRegistry.name_to_hash` to look one up by
name, call it to obtain a {class}`~.multihash.Hasher`, and feed it data to produce a
{class}`~.multihash.Digest`.

Example code:

```python
from sansio_tools.multihash import registry

# Compute a sha2-256 multihash.
hasher = registry.name_to_hash["sha2-256"]()
hasher.update(b"hello world")
digest = hasher.digest()

print(digest.to_multihash_bytes().hex())
# 1220b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9

print(digest.to_multihash_base64url())
# uEiC5TSe5k00-CKUuUtfafav6xITv43pTgO6QiPes4u_N6Q

# Decode a multihash (bytes or multibase base64url) back into a Digest.
decoded = registry.decode(digest.to_multihash_base64url())
assert decoded.digest == digest.digest
assert decoded.function.name == "sha2-256"
```

The module also exposes more low-level functions:

- {func}`~.multihash.multihash_varint_encode()` for the LEB128 varints used by the format, and the
  sans-io reader generators {func}`~.multihash.read_multihash_varint()`,
  {func}`~.multihash.read_multihash_raw_header()` and {func}`~.multihash.read_multihash_raw()` for
  decoding varints and raw multihashes from a {class}`~.parser.BinaryParser`.
- {func}`~.multihash.multibase_encode_base64url()` / {func}`~.multihash.multibase_decode_base64url()` for
  multibase base64url strings (the `u` prefix).
- {meth}`~.multihash.HashFunctionRegistry.register()` to add additional hash functions at runtime.
- {meth}`~.multihash.HashFunctionRegistry.read_from_binary_parser()` to read a multihash from a
  {class}`~.parser.BinaryParser` as a {class}`~.multihash.Digest`, or
  {func}`~.multihash.read_multihash_raw()` to split it into its `(function_code, digest)` components
  without requiring a registered function.
