Port tl-types fromm grammers

client/src/telethon/_impl/tl/__init__.py

@@ -0,0 +1,4 @@
+from . import abcs, core, functions, mtproto, types
+from .layer import LAYER, TYPE_MAPPING
+
+__all__ = ["abcs", "core", "functions", "mtproto", "types", "LAYER", "TYPE_MAPPING"]

client/src/telethon/_impl/tl/core/__init__.py

@@ -0,0 +1,5 @@
+from .reader import Reader
+from .request import Request
+from .serializable import Serializable, serialize_bytes_to
+
+__all__ = ["Reader", "Request", "Serializable", "serialize_bytes_to"]

client/src/telethon/_impl/tl/core/reader.py

@@ -0,0 +1,64 @@
+import struct
+from typing import TYPE_CHECKING, Any, Type, TypeVar
+
+if TYPE_CHECKING:
+    from .serializable import Serializable
+
+
+T = TypeVar("T", bound="Serializable")
+
+
+class Reader:
+    __slots__ = ("_buffer", "_pos", "_view")
+
+    def __init__(self, buffer: bytes) -> None:
+        self._buffer = buffer
+        self._pos = 0
+        self._view = memoryview(self._buffer)
+
+    def read(self, n: int) -> bytes:
+        self._pos += n
+        return self._view[self._pos - n : n]
+
+    def read_fmt(self, fmt: str, size: int) -> tuple[Any, ...]:
+        assert struct.calcsize(fmt) == size
+        self._pos += size
+        return struct.unpack(fmt, self._view[self._pos - size : self._pos])
+
+    def read_bytes(self) -> bytes:
+        if self._buffer[self._pos] == 254:
+            self._pos += 4
+            (length,) = struct.unpack(
+                "<i", self._buffer[self._pos - 3 : self._pos] + b"\0"
+            )
+            padding = length % 4
+        else:
+            length = self._buffer[self._pos]
+            padding = (length + 1) % 4
+            self._pos += 1
+
+        self._pos += length
+        data = self._view[self._pos - length : self._pos]
+        if padding > 0:
+            self._pos += 4 - padding
+
+        return data
+
+    @staticmethod
+    def _get_ty(_: int) -> Type["Serializable"]:
+        # Implementation replaced during import to prevent cycles,
+        # without the performance hit of having the import inside.
+        raise NotImplementedError
+
+    def read_serializable(self, cls: Type[T]) -> T:
+        # Calls to this method likely need to ignore "type-abstract".
+        # See https://github.com/python/mypy/issues/4717.
+        # Unfortunately `typing.cast` would add a tiny amount of runtime overhead
+        # which cannot be removed with optimization enabled.
+        self._pos += 4
+        cid = struct.unpack("<I", self._view[self._pos - 4 : self._pos])[0]
+        ty = self._get_ty(cid)
+        if ty is None:
+            raise ValueError(f"No type found for constructor ID: {cid:x}")
+        assert issubclass(ty, cls)
+        return ty._read_from(self)

client/src/telethon/_impl/tl/core/request.py

@@ -0,0 +1,20 @@
+import struct
+
+
+class Request:
+    __slots__ = "_body"
+
+    def __init__(self, body: bytes):
+        self._body = body
+
+    @property
+    def constructor_id(self) -> int:
+        try:
+            cid = struct.unpack("<i", self._body[:4])[0]
+            assert isinstance(cid, int)
+            return cid
+        except struct.error:
+            return 0
+
+    def debug_name(self) -> str:
+        return f"request#{self.constructor_id:x}"

client/src/telethon/_impl/tl/core/serializable.py

@@ -0,0 +1,52 @@
+import abc
+import struct
+from typing import Self, Tuple
+
+from .reader import Reader
+
+
+class Serializable(abc.ABC):
+    __slots__: Tuple[str, ...] = ()
+
+    @classmethod
+    @abc.abstractmethod
+    def constructor_id(cls) -> int:
+        pass
+
+    @classmethod
+    def _read_from(cls, reader: Reader) -> Self:
+        return reader.read_serializable(cls)
+
+    def _write_boxed_to(self, buffer: bytearray) -> None:
+        buffer += struct.pack("<I", self.constructor_id())
+        self._write_to(buffer)
+
+    @abc.abstractmethod
+    def _write_to(self, buffer: bytearray) -> None:
+        pass
+
+    @classmethod
+    def from_bytes(cls, blob: bytes) -> Self:
+        return Reader(blob).read_serializable(cls)
+
+    def __bytes__(self) -> bytes:
+        buffer = bytearray()
+        self._write_boxed_to(buffer)
+        return bytes(buffer)
+
+    def __repr__(self) -> str:
+        attrs = ", ".join(repr(getattr(self, attr)) for attr in self.__slots__)
+        return f"{self.__class__.__name__}({attrs})"
+
+
+def serialize_bytes_to(buffer: bytearray, data: bytes) -> None:
+    length = len(data)
+    if length < 0xFE:
+        buffer += struct.pack("<B", length)
+        length += 1
+    else:
+        buffer += b"\xfe"
+        buffer += struct.pack("<i", length)[:-1]
+
+    buffer += data
+    buffer += bytes((4 - (length % 4)) % 4)

client/src/telethon/_impl/tl/mtproto/core.py

@@ -0,0 +1 @@
+from ..core import *