Skip to content

Improve type hinting #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
qasikfwn wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Improve type hinting #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c2898f2
Add py.typed file
qasikfwn Jul 21, 2026
df7aad1
Fully type hint library
qasikfwn Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 65 additions & 57 deletions sjson/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,34 +13,54 @@

__version__ = '2.2.0'

_SUPPORTED_ENCODE_TYPE: typing.TypeAlias = typing.Union[
None,
bool,
numbers.Number,
str,
typing.Sequence['_SUPPORTED_ENCODE_TYPE'],
typing.Mapping[str, '_SUPPORTED_ENCODE_TYPE'],
]

_SUPPORTED_DECODE_TYPE: typing.TypeAlias = typing.Union[
None,
bool,
int,
float,
str,
list['_SUPPORTED_DECODE_TYPE'],
typing.OrderedDict[str, '_SUPPORTED_DECODE_TYPE'],
]

Loc = typing.NamedTuple('loc', [('line', int), ('column', int)])

class _InputStream:
@abstractmethod
def read(self, count: int = 1) -> bytes: ...

@abstractmethod
def skip(self, count: int = 1): ...
def skip(self, count: int = 1) -> None: ...

@abstractmethod
def peek(self, count: int = 1, allow_end_of_stream=False) -> bytes | None:
def peek(self, count: int = 1, allow_end_of_stream: bool=False) -> bytes | None:
pass

@abstractmethod
def get_location(self) -> tuple[int, int]: ...
def get_location(self) -> Loc: ...


class MemoryInputStream(_InputStream):
"""Input stream wrapper for reading directly from memory."""

def __init__(self, s: bytes):
def __init__(self, s: bytes) -> None:
"""
s -- a bytes object.
"""
self._stream = s
self._current_index = 0
self._length = len(s)

def read(self, count=1) -> bytes:
def read(self, count: int=1) -> bytes:
"""read ``count`` bytes from the stream."""
end_index = self._current_index + count
if end_index > self._length:
Expand All @@ -49,7 +69,7 @@ def read(self, count=1) -> bytes:
self._current_index = end_index
return result

def peek(self, count=1, allow_end_of_stream=False) -> bytes | None:
def peek(self, count: int=1, allow_end_of_stream: bool=False) -> bytes | None:
"""peek ``count`` bytes from the stream. If ``allow_end_of_stream`` is
``True``, no error will be raised if the end of the stream is reached
while trying to peek."""
Expand All @@ -61,13 +81,12 @@ def peek(self, count=1, allow_end_of_stream=False) -> bytes | None:

return self._stream[self._current_index : end_index]

def skip(self, count=1):
def skip(self, count: int=1) -> None:
"""skip ``count`` bytes."""
self._current_index += count

def get_location(self) -> tuple[int, int]:
def get_location(self) -> Loc:
"""Get the current location in the stream."""
loc = collections.namedtuple('loc', ['line', 'column'])
bytes_read = self._stream[: self._current_index]
line = 1
column = 1
Expand All @@ -78,19 +97,19 @@ def get_location(self) -> tuple[int, int]:
column = 1
else:
column += 1
return loc(line, column)
return Loc(line, column)


class ByteBufferInputStream(_InputStream):
"""Input stream wrapper for reading directly from an I/O object."""

def __init__(self, stream: io.BufferedReader):
def __init__(self, stream: io.BufferedReader) -> None:
self._stream = stream
self._index = 0
self._line = 1
self._column = 1

def read(self, count=1) -> bytes:
def read(self, count: int=1) -> bytes:
"""read ``count`` bytes from the stream."""
result = self._stream.read(count)
if len(result) < count:
Expand All @@ -105,7 +124,7 @@ def read(self, count=1) -> bytes:
self._column += 1
return result

def peek(self, count=1, allow_end_of_stream=False) -> bytes | None:
def peek(self, count: int=1, allow_end_of_stream: bool=False) -> bytes | None:
"""peek ``count`` bytes from the stream. If ``allow_end_of_stream`` is
``True``, no error will be raised if the end of the stream is reached
while trying to peek."""
Expand All @@ -117,29 +136,28 @@ def peek(self, count=1, allow_end_of_stream=False) -> bytes | None:

return result[:count]

def skip(self, count=1):
def skip(self, count: int=1) -> None:
"""skip ``count`` bytes."""
self.read(count)

def get_location(self) -> tuple[int, int]:
def get_location(self) -> Loc:
"""Get the current location in the stream."""
loc = collections.namedtuple('loc', ['line', 'column'])
return loc(self._line, self._column)
return Loc(self._line, self._column)


class ParseException(RuntimeError):
"""Parse exception."""

def __init__(self, msg: str, location: tuple[int, int]):
def __init__(self, msg: str, location: Loc) -> None:
super(ParseException, self).__init__(msg)
self._msg = msg
self._location = location

def get_location(self) -> tuple[int, int]:
def get_location(self) -> Loc:
"""Get the current location at which the exception occurred."""
return self._location

def __str__(self):
def __str__(self) -> str:
return '{} at line {}, column {}'.format(
self._msg, self._location.line, self._location.column
)
Expand All @@ -149,7 +167,7 @@ def _raise_end_of_stream_exception(stream: _InputStream) -> typing.NoReturn:
raise ParseException('Unexpected end-of-stream', stream.get_location())


def _consume(stream: _InputStream, what: bytes):
def _consume(stream: _InputStream, what: bytes) -> None:
_skip_whitespace(stream)
what_len = len(what)
if stream.peek(what_len) != what:
Expand All @@ -159,7 +177,7 @@ def _consume(stream: _InputStream, what: bytes):
stream.skip(what_len)


def _skip_characters_and_whitespace(stream: _InputStream, num_char_to_skip: int):
def _skip_characters_and_whitespace(stream: _InputStream, num_char_to_skip: int) -> bytes | None:
stream.skip(num_char_to_skip)
return _skip_whitespace(stream)

Expand All @@ -169,11 +187,11 @@ def _skip_characters_and_whitespace(stream: _InputStream, num_char_to_skip: int)
_WHITESPACE_SET = frozenset([b' ', b'\t', b'\n', b'\r'])


def _is_whitespace(char: bytes | None):
def _is_whitespace(char: bytes | None) -> bool:
return char in _WHITESPACE_SET


def _skip_c_style_comment(stream: _InputStream):
def _skip_c_style_comment(stream: _InputStream) -> None:
comment_start_location = stream.get_location()
# skip the comment start
stream.skip(2)
Expand All @@ -196,7 +214,7 @@ def _skip_c_style_comment(stream: _InputStream):
stream.skip()


def _skip_cpp_style_comment(stream: _InputStream):
def _skip_cpp_style_comment(stream: _InputStream) -> None:
# skip the comment start
stream.skip(2)
while True:
Expand All @@ -206,7 +224,7 @@ def _skip_cpp_style_comment(stream: _InputStream):
stream.skip()


def _skip_whitespace(stream: _InputStream):
def _skip_whitespace(stream: _InputStream) -> bytes | None:
"""skip whitespace. Returns the next character if a new position within the
stream was found; returns None if the end of the stream was hit."""
while True:
Expand All @@ -230,11 +248,11 @@ def _skip_whitespace(stream: _InputStream):
_IDENTIFIER_SET = frozenset(string.ascii_letters + string.digits + '_')


def _is_identifier(obj: bytes):
def _is_identifier(obj: bytes) -> bool:
return chr(obj[0]) in _IDENTIFIER_SET


def _decode_escaped_character(char: bytes):
def _decode_escaped_character(char: bytes) -> bytes:
match char:
case b'b':
return b'\b'
Expand All @@ -256,7 +274,7 @@ class RawQuoteStyle(Enum):
Python = 2


def _decode_string(stream: _InputStream, allow_identifier=False):
def _decode_string(stream: _InputStream, allow_identifier: bool=False) -> str:
# When we enter here, we either start with " or [, or there is no quoting
# enabled.
_skip_whitespace(stream)
Expand Down Expand Up @@ -357,7 +375,7 @@ def _decode_string(stream: _InputStream, allow_identifier=False):
_NUMBER_SEPARATOR_SET = _WHITESPACE_SET.union({b',', b']', b'}', None})


def _decode_number(stream: _InputStream, next_char):
def _decode_number(stream: _InputStream, next_char: bytes) -> float | int:
"""Parse a number.

next_char -- the next byte in the stream.
Expand All @@ -384,14 +402,14 @@ def _decode_number(stream: _InputStream, next_char):
return int(value)


def _decode_dict(stream: _InputStream, delimited=False):
def _decode_dict(stream: _InputStream, delimited: bool=False) -> typing.OrderedDict[str, _SUPPORTED_DECODE_TYPE]:
"""
delimited -- if ``True``, parsing will stop once the end-of-dictionary
delimiter has been reached(``}``)
"""
from collections import OrderedDict

result = OrderedDict()
result: _SUPPORTED_DECODE_TYPE = OrderedDict()

if stream.peek() == b'{':
stream.skip()
Expand Down Expand Up @@ -421,8 +439,8 @@ def _decode_dict(stream: _InputStream, delimited=False):
return result


def _parse_list(stream: _InputStream):
result = []
def _parse_list(stream: _InputStream) -> list[_SUPPORTED_DECODE_TYPE]:
result: _SUPPORTED_DECODE_TYPE = []
# skip '['
next_char = _skip_characters_and_whitespace(stream, 1)

Expand All @@ -441,7 +459,7 @@ def _parse_list(stream: _InputStream):
return result


def _parse(stream: _InputStream):
def _parse(stream: _InputStream) -> _SUPPORTED_DECODE_TYPE:
next_char = _skip_whitespace(stream)

match next_char:
Expand Down Expand Up @@ -475,20 +493,20 @@ def _parse(stream: _InputStream):
raise ParseException('Invalid character', stream.get_location())


def load(stream: io.RawIOBase):
def load(stream: io.RawIOBase) -> typing.OrderedDict[str, _SUPPORTED_DECODE_TYPE]:
"""Load a SJSON object from a stream.

The stream is assumed to point to UTF-8 encoded data.
"""
return _decode_dict(ByteBufferInputStream(io.BufferedReader(stream)))


def loads(text: str):
def loads(text: str) -> typing.OrderedDict[str, _SUPPORTED_DECODE_TYPE]:
"""Load a SJSON object from a string."""
return _decode_dict(MemoryInputStream(text.encode('utf-8')))


def dumps(obj, indent=None) -> str:
def dumps(obj: _SUPPORTED_ENCODE_TYPE, indent: int | str | None=None) -> str:
"""Dump an object to a string."""
import io

Expand All @@ -497,7 +515,7 @@ def dumps(obj, indent=None) -> str:
return stream.getvalue()


def dump(obj, fp: io.TextIOBase, indent: int | None | str = None):
def dump(obj: _SUPPORTED_ENCODE_TYPE, fp: io.TextIOBase, indent: int | None | str = None) -> None:
"""Dump an object to a text stream.

The output is always text, not binary."""
Expand All @@ -517,7 +535,7 @@ def dump(obj, fp: io.TextIOBase, indent: int | None | str = None):
_ESCAPE_CHARACTER_SET = {'\n': '\\n', '\b': '\\b', '\t': '\\t', '"': '\\"'}


def _escape_string(obj: str, quote=True) -> typing.Generator[str, None, None]:
def _escape_string(obj: str, quote: bool=True) -> typing.Generator[str, None, None]:
"""Escape a string.

If quote is set, the string will be returned with quotation marks at the
Expand All @@ -539,22 +557,12 @@ def _escape_string(obj: str, quote=True) -> typing.Generator[str, None, None]:
yield '"'


_SUPPORTED_ENCODE_TYPE: typing.TypeAlias = typing.Union[
None,
bool,
numbers.Number,
str,
typing.Sequence['_SUPPORTED_ENCODE_TYPE'],
typing.Mapping[str, '_SUPPORTED_ENCODE_TYPE'],
]


def _encode(
obj: _SUPPORTED_ENCODE_TYPE,
separators: tuple[str, str, str] = (', ', '\n', ' = '),
indent: str = '',
level: int = 0,
):
) -> typing.Generator[str, None, None]:
match obj:
case None:
yield 'null'
Expand All @@ -577,18 +585,18 @@ def _encode(
raise RuntimeError("Unsupported object type")


def _indent(level: int, indent: str):
def _indent(level: int, indent: str) -> str:
return indent * level


def _encode_key(k: str):
def _encode_key(k: str) -> typing.Generator[str, None, None]:
yield from _escape_string(k, False)


def _encode_list(
obj: typing.Sequence, separators: tuple[str, str, str],
obj: typing.Sequence['_SUPPORTED_ENCODE_TYPE'], separators: tuple[str, str, str],
indent: str, level: int
):
) -> typing.Generator[str, None, None]:
yield '['
first = True
for element in obj:
Expand All @@ -601,9 +609,9 @@ def _encode_list(


def _encode_dict(
obj: typing.Mapping, separators: tuple[str, str, str],
obj: typing.Mapping[str, _SUPPORTED_ENCODE_TYPE], separators: tuple[str, str, str],
indent: str, level: int
):
) -> typing.Generator[str, None, None]:
if level > 0:
yield '{'
yield separators[1]
Expand Down
Empty file added sjson/py.typed
View file
Open in desktop
Empty file.