Metadata-Version: 2.4
Name: sansio_tools
Version: 1.1.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)

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

## {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.
