Serialization and parsing¶

Serial¶

kio.serial.entity_reader(entity_type: type[E], nullable: bool = False) → Reader[E | None][source]¶

Parameters:

entity_type (type[TypeVar(E, bound= Entity)])
nullable (bool)

Return type:

Reader[TypeVar(E, bound= Entity) | None]

kio.serial.entity_writer(entity_type: type[E], nullable: bool = False) → Callable[[StreamWriter | IO[bytes], E | None], None][source]¶

Parameters:

entity_type (type[TypeVar(E, bound= Entity)])
nullable (bool)

Return type:

Callable[[StreamWriter | IO[bytes], TypeVar(E, bound= Entity) | None], None]

Readers¶

class kio.serial.readers.Reader(*args, **kwargs)[source]¶: Bases: Protocol[T_co]

kio.serial.readers.read_boolean(value_bytes: Buffer) → bool[source]¶

Parameters:: value_bytes (Buffer)
Return type:: bool

kio.serial.readers.read_int8(value_bytes: Buffer, /) → i8[source]¶

Parameters:: value_bytes (Buffer)
Return type:: i8

kio.serial.readers.read_int16(value_bytes: Buffer, /) → i16[source]¶

Parameters:: value_bytes (Buffer)
Return type:: i16

kio.serial.readers.read_int32(value_bytes: Buffer, /) → i32[source]¶

Parameters:: value_bytes (Buffer)
Return type:: i32

kio.serial.readers.read_int64(value_bytes: Buffer) → i64[source]¶

Parameters:: value_bytes (Buffer)
Return type:: i64

kio.serial.readers.read_uint8(value_bytes: Buffer) → u8[source]¶

Parameters:: value_bytes (Buffer)
Return type:: u8

kio.serial.readers.read_uint16(value_bytes: Buffer) → u16[source]¶

Parameters:: value_bytes (Buffer)
Return type:: u16

kio.serial.readers.read_uint32(value_bytes: Buffer) → u32[source]¶

Parameters:: value_bytes (Buffer)
Return type:: u32

kio.serial.readers.read_uint64(value_bytes: Buffer) → u64[source]¶

Parameters:: value_bytes (Buffer)
Return type:: u64

kio.serial.readers.read_unsigned_varint(buffer: Buffer, offset: int, /, _max_bytes: int = 5) → tuple[int, int][source]¶

Deserialize an integer stored into variable number of bytes (1-5).

Parameters:

buffer (Buffer)
offset (int)
_max_bytes (int)

Return type:

tuple[int, int]

kio.serial.readers.read_signed_varint(buffer: Buffer, offset: int) → tuple[svarint, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[svarint, int]

kio.serial.readers.read_unsigned_varlong(buffer: Buffer, offset: int) → tuple[uvarlong, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[uvarlong, int]

kio.serial.readers.read_signed_varlong(buffer: Buffer, offset: int) → tuple[svarlong, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[svarlong, int]

kio.serial.readers.read_float64(value_bytes: Buffer) → float[source]¶

Parameters:: value_bytes (Buffer)
Return type:: float

kio.serial.readers.read_compact_string_as_bytes(buffer: Buffer, offset: int, /) → tuple[bytes, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[bytes, int]

kio.serial.readers.read_compact_string_as_bytes_nullable(buffer: Buffer, offset: int, /) → tuple[bytes | None, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[bytes | None, int]

kio.serial.readers.read_compact_string(buffer: Buffer, offset: int, /) → tuple[str, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[str, int]

kio.serial.readers.read_compact_string_nullable(buffer: Buffer, offset: int, /) → tuple[str | None, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[str | None, int]

kio.serial.readers.read_legacy_bytes(buffer: Buffer, offset: int, /) → tuple[bytes, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[bytes, int]

kio.serial.readers.read_nullable_legacy_bytes(buffer: Buffer, offset: int, /) → tuple[bytes | None, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[bytes | None, int]

kio.serial.readers.read_legacy_string(buffer: Buffer, offset: int, /) → tuple[str, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[str, int]

kio.serial.readers.read_nullable_legacy_string(buffer: Buffer, offset: int, /) → tuple[str | None, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[str | None, int]

kio.serial.readers.read_legacy_array_length(value_bytes: Buffer, /) → i32¶

Parameters:: value_bytes (Buffer)
Return type:: i32

kio.serial.readers.read_compact_array_length(buffer: Buffer, offset: int, /) → tuple[int | None, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[int | None, int]

kio.serial.readers.read_uuid(value_bytes: Buffer) → UUID | None[source]¶

Parameters:: value_bytes (Buffer)
Return type:: UUID | None

kio.serial.readers.compact_array_reader(item_reader: Reader[T]) → Reader[tuple[T, ...] | None][source]¶

Parameters:: item_reader (Reader[TypeVar(T)])
Return type:: Reader[tuple[TypeVar(T), ...] | None]

kio.serial.readers.legacy_array_reader(item_reader: Reader[T]) → Reader[tuple[T, ...] | None][source]¶

Parameters:: item_reader (Reader[TypeVar(T)])
Return type:: Reader[tuple[TypeVar(T), ...] | None]

kio.serial.readers.read_error_code(buffer: Buffer, offset: int, /) → tuple[ErrorCode, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[ErrorCode, int]

kio.serial.readers.read_timedelta_i32(buffer: Buffer, offset: int, /) → tuple[i32Timedelta, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[i32Timedelta, int]

kio.serial.readers.read_timedelta_i64(buffer: Buffer, offset: int, /) → tuple[i64Timedelta, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[i64Timedelta, int]

kio.serial.readers.tz_aware_from_i64(timestamp: i64) → TZAware[source]¶

Parameters:: timestamp (i64)
Return type:: TZAware

kio.serial.readers.read_datetime_i64(buffer: Buffer, offset: int, /) → tuple[TZAware, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[TZAware, int]

kio.serial.readers.read_nullable_datetime_i64(buffer: Buffer, offset: int, /) → tuple[TZAware | None, int][source]¶

Parameters:

buffer (Buffer)
offset (int)

Return type:

tuple[TZAware | None, int]

Writers¶

kio.serial.writers.write_boolean(buffer: StreamWriter | IO[bytes], value: bool) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (bool)

Return type:

None

kio.serial.writers.write_int8(buffer: StreamWriter | IO[bytes], value: i8) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (i8)

Return type:

None

kio.serial.writers.write_int16(buffer: StreamWriter | IO[bytes], value: i16) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (i16)

Return type:

None

kio.serial.writers.write_int32(buffer: StreamWriter | IO[bytes], value: i32) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (i32)

Return type:

None

kio.serial.writers.write_int64(buffer: StreamWriter | IO[bytes], value: i64) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (i64)

Return type:

None

kio.serial.writers.write_uint8(buffer: StreamWriter | IO[bytes], value: u8) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (u8)

Return type:

None

kio.serial.writers.write_uint16(buffer: StreamWriter | IO[bytes], value: u16) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (u16)

Return type:

None

kio.serial.writers.write_uint32(buffer: StreamWriter | IO[bytes], value: u32) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (u32)

Return type:

None

kio.serial.writers.write_uint64(buffer: StreamWriter | IO[bytes], value: u64) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (u64)

Return type:

None

kio.serial.writers.write_unsigned_varint(buffer: StreamWriter | IO[bytes], value: uvarint) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (uvarint)

Return type:

None

kio.serial.writers.write_unsigned_varlong(buffer: StreamWriter | IO[bytes], value: int) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (int)

Return type:

None

kio.serial.writers.write_signed_varint(buffer: StreamWriter | IO[bytes], value: int) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (int)

Return type:

None

kio.serial.writers.write_signed_varlong(buffer: StreamWriter | IO[bytes], value: int) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (int)

Return type:

None

kio.serial.writers.write_float64(buffer: StreamWriter | IO[bytes], value: float) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (float)

Return type:

None

kio.serial.writers.write_nullable_compact_string(buffer: StreamWriter | IO[bytes], value: str | bytes | None) → None[source]¶

Write a nullable string with compact length encoding.

Parameters:

buffer (StreamWriter | IO[bytes])
value (str | bytes | None)

Return type:

None

kio.serial.writers.write_compact_string(buffer: StreamWriter | IO[bytes], value: str | bytes) → None[source]¶

Write a non-nullable string with compact length encoding.

Parameters:

buffer (StreamWriter | IO[bytes])
value (str | bytes)

Return type:

None

kio.serial.writers.write_nullable_legacy_string(buffer: StreamWriter | IO[bytes], value: str | None) → None[source]¶

Write a nullable string with legacy int16 length encoding.

Parameters:

buffer (StreamWriter | IO[bytes])
value (str | None)

Return type:

None

kio.serial.writers.write_nullable_legacy_bytes(buffer: StreamWriter | IO[bytes], value: bytes | None) → None[source]¶

Write a nullable bytes with legacy int32 length encoding.

Parameters:

buffer (StreamWriter | IO[bytes])
value (bytes | None)

Return type:

None

kio.serial.writers.write_legacy_string(buffer: StreamWriter | IO[bytes], value: str) → None[source]¶

Write a non-nullable string with legacy int16 length encoding.

Parameters:

buffer (StreamWriter | IO[bytes])
value (str)

Return type:

None

kio.serial.writers.write_legacy_bytes(buffer: StreamWriter | IO[bytes], value: bytes) → None[source]¶

Write a non-nullable bytes with legacy int32 length encoding.

Parameters:

buffer (StreamWriter | IO[bytes])
value (bytes)

Return type:

None

kio.serial.writers.write_empty_tagged_fields(buffer: StreamWriter | IO[bytes]) → None[source]¶

Parameters:: buffer (StreamWriter | IO[bytes])
Return type:: None

kio.serial.writers.write_legacy_array_length(buffer: StreamWriter | IO[bytes], value: i32) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (i32)

Return type:

None

kio.serial.writers.write_compact_array_length(buffer: StreamWriter | IO[bytes], value: int | None) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (int | None)

Return type:

None

kio.serial.writers.write_uuid(buffer: StreamWriter | IO[bytes], value: UUID | None) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (UUID | None)

Return type:

None

kio.serial.writers.compact_array_writer(item_writer: Callable[[StreamWriter | IO[bytes], T], None]) → Callable[[StreamWriter | IO[bytes], Sequence[T] | None], None][source]¶

Parameters:: item_writer (Callable[[StreamWriter | IO[bytes], TypeVar(T)], None])
Return type:: Callable[[StreamWriter | IO[bytes], Sequence[TypeVar(T)] | None], None]

kio.serial.writers.legacy_array_writer(item_writer: Callable[[StreamWriter | IO[bytes], T], None]) → Callable[[StreamWriter | IO[bytes], Sequence[T] | None], None][source]¶

Parameters:: item_writer (Callable[[StreamWriter | IO[bytes], TypeVar(T)], None])
Return type:: Callable[[StreamWriter | IO[bytes], Sequence[TypeVar(T)] | None], None]

kio.serial.writers.write_tagged_field(buffer: StreamWriter | IO[bytes], tag: uvarint, writer: Callable[[StreamWriter | IO[bytes], T], None], value: T) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
tag (uvarint)
writer (Callable[[StreamWriter | IO[bytes], TypeVar(T)], None])
value (TypeVar(T))

Return type:

None

kio.serial.writers.write_error_code(buffer: StreamWriter | IO[bytes], error_code: ErrorCode) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
error_code (ErrorCode)

Return type:

None

kio.serial.writers.write_timedelta_i32(buffer: StreamWriter | IO[bytes], value: i32Timedelta) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (i32Timedelta)

Return type:

None

kio.serial.writers.write_timedelta_i64(buffer: StreamWriter | IO[bytes], value: i64Timedelta) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (i64Timedelta)

Return type:

None

kio.serial.writers.write_datetime_i64(buffer: StreamWriter | IO[bytes], value: datetime) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (datetime)

Return type:

None

kio.serial.writers.write_nullable_datetime_i64(buffer: StreamWriter | IO[bytes], value: datetime | None) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
value (datetime | None)

Return type:

None

Errors¶

exception kio.serial.errors.SerialError[source]¶: Bases: Exception

exception kio.serial.errors.DecodeError[source]¶: Bases: SerialError

exception kio.serial.errors.UnexpectedNull[source]¶: Bases: DecodeError

exception kio.serial.errors.BufferUnderflow[source]¶: Bases: DecodeError

exception kio.serial.errors.SchemaError[source]¶: Bases: SerialError

exception kio.serial.errors.OutOfBoundValue[source]¶: Bases: SerialError

exception kio.serial.errors.EncodeError[source]¶: Bases: SerialError