Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
# Changelog

## [UNRELEASED]
**New features:**
- Added `Subconstruct`, `SymmetricAdapter`, `Tunnel` and `Validator` to `construct_typed` as subscriptable types.

## [0.8.0] - 2026-07-20
**Breaking changes:**
- `csfield` should only be used for constructs that cannot build from `None`. Every other construct should use the new `csfield_noinit`, `csfield_const` or `csfield_default`.
Expand Down
8 changes: 8 additions & 0 deletions construct_typed/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,10 @@
Context,
ListContainer,
PathType,
Subconstruct,
SymmetricAdapter,
Tunnel,
Validator,
)
from .tenum import EnumBase, EnumValue, FlagsEnumBase, TEnum, TFlagsEnum

Expand All @@ -43,11 +47,15 @@
"FlagsEnumBase",
"TEnum",
"TFlagsEnum",
"Subconstruct",
"Adapter",
"ConstantOrContextLambda",
"Construct",
"Context",
"ListContainer",
"PathType",
"Array",
"SymmetricAdapter",
"Tunnel",
"Validator",
]
42 changes: 35 additions & 7 deletions construct_typed/generic_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,10 @@
from construct import Array as Array
from construct import Construct as Construct
from construct import ListContainer as ListContainer
from construct import Subconstruct as Subconstruct
from construct import SymmetricAdapter as SymmetricAdapter
from construct import Tunnel as Tunnel
from construct import Validator as Validator
from construct.core import ConstantOrContextLambda as ConstantOrContextLambda
from construct.core import Context as Context
from construct.core import PathType as PathType
Expand All @@ -22,26 +26,50 @@
else:
import construct as cs

# at runtime, the original classes are no generics, so whe have to make new classes with generics support
# at runtime, the original classes are not generic, so we have to make new classes with generics support
class Construct(t.Generic[ParsedType, BuildTypes], cs.Construct):
pass
"""Subscriptable version of `construct.Construct` that can be used as a generic type."""

class Subconstruct(
t.Generic[SubconParsedType, SubconBuildTypes, ParsedType, BuildTypes],
cs.Subconstruct,
):
"""Subscriptable version of `construct.Subconstruct` that can be used as a generic type."""

class Adapter(
t.Generic[SubconParsedType, SubconBuildTypes, ParsedType, BuildTypes],
cs.Adapter,
):
pass
"""Subscriptable version of `construct.Adapter` that can be used as a generic type."""

class ListContainer(t.Generic[ListType], cs.ListContainer):
pass
class Tunnel(
t.Generic[SubconParsedType, SubconBuildTypes],
cs.Tunnel,
):
"""Subscriptable version of `construct.Tunnel` that can be used as a generic type."""

class Context:
pass
class SymmetricAdapter(
t.Generic[SubconParsedType, SubconBuildTypes, ParsedType, BuildTypes],
cs.SymmetricAdapter,
):
"""Subscriptable version of `construct.SymmetricAdapter` that can be used as a generic type."""

class Validator(
t.Generic[SubconParsedType, SubconBuildTypes],
cs.Validator,
):
"""Subscriptable version of `construct.Validator` that can be used as a generic type."""

class Array(
t.Generic[SubconParsedType, SubconBuildTypes],
cs.Array,
):
"""Subscriptable version of `construct.Array` that can be used as a generic type."""

class ListContainer(t.Generic[ListType], cs.ListContainer):
pass

class Context:
pass

ConstantOrContextLambda = ValueType | t.Callable[[Context], t.Any]
Expand Down
116 changes: 113 additions & 3 deletions tests/test_typed.py
Original file line number Diff line number Diff line change
Expand Up @@ -203,9 +203,7 @@ def test_dataclass_ifthenelse() -> None:
@dataclasses.dataclass
class IfThenElseTest(DataclassMixin):
test_if: int | None = csfield(cs.If(False, cs.Int8ub))
test_ifthenelse: int | None = csfield(
cs.IfThenElse(True, cs.Int8ub, cs.Pass)
)
test_ifthenelse: int | None = csfield(cs.IfThenElse(True, cs.Int8ub, cs.Pass))

a = IfThenElseTest(test_if=None, test_ifthenelse=None)
assert a.test_if is None
Expand Down Expand Up @@ -745,3 +743,115 @@ class TestEnum(cst.FlagsEnumBase):
assert TestEnum.Value_NoDoc.__doc__ == ""
assert TestEnum.Value_NoDoc2.__doc__ == ""
assert TestEnum(8).__doc__ == "missing value"


def test_construct_subscriptable() -> None:
"""Construct has to be subscriptable at runtime."""

class MyByte(cst.Construct[int, int]):
def _parse(
self, stream: t.IO[bytes], context: cst.Context, path: cst.PathType
) -> int:
return cs.stream_read(stream, 1, path)[0]

def _build(
self,
obj: int,
stream: t.IO[bytes],
context: cst.Context,
path: cst.PathType,
) -> int:
cs.stream_write(stream, bytes([obj]), 1, path)
return obj

def _sizeof(self, context: cst.Context, path: cst.PathType) -> int:
return 1

fmt = MyByte()
assert fmt.parse(b"\x05") == 5
assert fmt.build(5) == b"\x05"
assert fmt.sizeof() == 1


def test_subconstruct_subscriptable() -> None:
"""Subconstruct has to be subscriptable at runtime."""

class MyPaddedBytes(cst.Subconstruct[bytes, bytes, bytes, bytes]):
def __init__(self) -> None:
pass

fmt = MyPaddedBytes()
assert fmt is not None


def test_adapter_subscriptable() -> None:
"""Adapter has to be subscriptable at runtime."""

class IpAddressAdapter(cst.Adapter[cst.ListContainer[int], list[int], str, str]):
def _encode(
self, obj: str, context: cst.Context, path: cst.PathType
) -> list[int]:
return list(map(int, obj.split(".")))

def _decode(
self, obj: list[int], context: cst.Context, path: cst.PathType
) -> str:
return "{0}.{1}.{2}.{3}".format(*obj)

IpAddress2 = IpAddressAdapter(cs.Byte[4])

assert IpAddress2.parse(b"\x7f\x80\x81\x82") == "127.128.129.130"
assert IpAddress2.build("127.1.2.3") == b"\x7f\x01\x02\x03"
assert IpAddress2.sizeof() == 4


def test_symmetric_adapter_subscriptable() -> None:
"""SymmetricAdapter has to be subscriptable at runtime."""

T = t.TypeVar("T")

class ReversedList(cst.SymmetricAdapter[list[T], list[T], list[T], list[T]]):
def _decode(
self, obj: list[T], context: cst.Context, path: cst.PathType
) -> list[T]:
return list(reversed(obj))

assert ReversedList(cs.Array(4, cs.Byte)).build([1, 2, 3, 4]) == b"\x04\x03\x02\x01"
assert ReversedList(cs.Array(4, cs.Byte)).parse(b"\x01\x02\x03\x04") == [4, 3, 2, 1]


def test_validator_subscriptable() -> None:
"""Validator has to be subscriptable at runtime."""

class OneOf(cst.Validator[int, int]):
def _validate(
self, obj: int, context: cst.Context, path: cst.PathType
) -> bool:
return obj in (1, 2, 3)

fmt = OneOf(cs.Byte)

assert fmt.parse(b"\x01") == 1
assert fmt.build(2) == b"\x02"
assert raises(fmt.build, 4) is cs.ValidationError
assert raises(fmt.parse, b"\x04") is cs.ValidationError


def test_tunnel_subscriptable() -> None:
"""Tunnel has to be subscriptable at runtime."""

class ROT1(cst.Tunnel[bytes, bytes]):
def _decode(
self, data: bytes, context: cst.Context, path: cst.PathType
) -> bytes:
return bytes((b - 1) % 256 for b in data)

def _encode(
self, data: bytes, context: cst.Context, path: cst.PathType
) -> bytes:
return bytes((b + 1) % 256 for b in data)

fmt = ROT1(cs.GreedyBytes)

assert fmt.parse(b"\x02\x03\x04") == b"\x01\x02\x03"
assert fmt.build(b"\x01\x02\x03") == b"\x02\x03\x04"