Serialization and parsing¶

Serial¶

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

Parameters:

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

Return type:

Callable[[IO[bytes]], 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¶

kio.serial.readers.read_exact(buffer: IO[bytes], num_bytes: int) → bytes[source]¶

Parameters:

buffer (IO[bytes])
num_bytes (int)

Return type:

bytes

kio.serial.readers.read_boolean(buffer: IO[bytes]) → bool[source]¶

Parameters:: buffer (IO[bytes])
Return type:: bool

kio.serial.readers.read_int8(buffer: IO[bytes]) → i8[source]¶

Parameters:: buffer (IO[bytes])
Return type:: i8

kio.serial.readers.read_int16(buffer: IO[bytes]) → i16[source]¶

Parameters:: buffer (IO[bytes])
Return type:: i16

kio.serial.readers.read_int32(buffer: IO[bytes]) → i32[source]¶

Parameters:: buffer (IO[bytes])
Return type:: i32

kio.serial.readers.read_int64(buffer: IO[bytes]) → i64[source]¶

Parameters:: buffer (IO[bytes])
Return type:: i64

kio.serial.readers.read_uint8(buffer: IO[bytes]) → u8[source]¶

Parameters:: buffer (IO[bytes])
Return type:: u8

kio.serial.readers.read_uint16(buffer: IO[bytes]) → u16[source]¶

Parameters:: buffer (IO[bytes])
Return type:: u16

kio.serial.readers.read_uint32(buffer: IO[bytes]) → u32[source]¶

Parameters:: buffer (IO[bytes])
Return type:: u32

kio.serial.readers.read_uint64(buffer: IO[bytes]) → u64[source]¶

Parameters:: buffer (IO[bytes])
Return type:: u64

kio.serial.readers.read_unsigned_varint(buffer: IO[bytes]) → int[source]¶

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

Parameters:: buffer (IO[bytes])
Return type:: int

kio.serial.readers.read_float64(buffer: IO[bytes]) → float[source]¶

Parameters:: buffer (IO[bytes])
Return type:: float

kio.serial.readers.read_compact_string_as_bytes(buffer: IO[bytes]) → bytes[source]¶

Parameters:: buffer (IO[bytes])
Return type:: bytes

kio.serial.readers.read_compact_string_as_bytes_nullable(buffer: IO[bytes]) → bytes | None[source]¶

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

kio.serial.readers.read_compact_string(buffer: IO[bytes]) → str[source]¶

Parameters:: buffer (IO[bytes])
Return type:: str

kio.serial.readers.read_compact_string_nullable(buffer: IO[bytes]) → str | None[source]¶

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

kio.serial.readers.read_legacy_bytes(buffer: IO[bytes]) → bytes[source]¶

Parameters:: buffer (IO[bytes])
Return type:: bytes

kio.serial.readers.read_nullable_legacy_bytes(buffer: IO[bytes]) → bytes | None[source]¶

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

kio.serial.readers.read_legacy_string(buffer: IO[bytes]) → str[source]¶

Parameters:: buffer (IO[bytes])
Return type:: str

kio.serial.readers.read_nullable_legacy_string(buffer: IO[bytes]) → str | None[source]¶

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

kio.serial.readers.read_legacy_array_length(buffer: IO[bytes]) → i32¶

Parameters:: buffer (IO[bytes])
Return type:: i32

kio.serial.readers.read_compact_array_length(buffer: IO[bytes]) → int[source]¶

Parameters:: buffer (IO[bytes])
Return type:: int

kio.serial.readers.read_uuid(buffer: IO[bytes]) → UUID | None[source]¶

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

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

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

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

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

kio.serial.readers.read_error_code(buffer: IO[bytes]) → ErrorCode[source]¶

Parameters:: buffer (IO[bytes])
Return type:: ErrorCode

kio.serial.readers.read_timedelta_i32(buffer: IO[bytes]) → i32Timedelta[source]¶

Parameters:: buffer (IO[bytes])
Return type:: i32Timedelta

kio.serial.readers.read_timedelta_i64(buffer: IO[bytes]) → i64Timedelta[source]¶

Parameters:: buffer (IO[bytes])
Return type:: i64Timedelta

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

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

kio.serial.readers.read_datetime_i64(buffer: IO[bytes]) → TZAware[source]¶

Parameters:: buffer (IO[bytes])
Return type:: TZAware

kio.serial.readers.read_nullable_datetime_i64(buffer: IO[bytes]) → TZAware | None[source]¶

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

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: int) → None[source]¶

Serialize an integer value between 0 and 2^31 - 1 to bytearray using 1-5 bytes depending on value size.

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[source]¶

Parameters:

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

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: int, writer: Callable[[StreamWriter | IO[bytes], T], None], value: T) → None[source]¶

Parameters:

buffer (StreamWriter | IO[bytes])
tag (int)
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