Skip to content

Encoding

The encoder transparently selects a storage strategy based on payload size, keeping callback data within Telegram's 64-byte limit.

CallbackEncoder

telegram_menu_builder.encoding.CallbackEncoder

Handles encoding and decoding of callback data with intelligent compression.

This class implements a hybrid approach: - Small data (< 60 bytes): Inline in callback_data - Medium data (60-500 bytes): Temporary storage with reference - Large data (> 500 bytes): Persistent storage with reference

Attributes:

Name Type Description
storage

Storage backend for non-inline data
prefix_inline

Prefix for inline data ('I:')
prefix_short

Prefix for short-term storage ('S:')
prefix_persistent

Prefix for persistent storage ('P:')
Example

encoder = CallbackEncoder(storage) action = MenuAction(handler="test", params={"id": 123}) encoded = await encoder.encode(action) decoded = await encoder.decode(encoded)

Source code in src/telegram_menu_builder/encoding.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
class CallbackEncoder:
    """Handles encoding and decoding of callback data with intelligent compression.

    This class implements a hybrid approach:
    - Small data (< 60 bytes): Inline in callback_data
    - Medium data (60-500 bytes): Temporary storage with reference
    - Large data (> 500 bytes): Persistent storage with reference

    Attributes:
        storage: Storage backend for non-inline data
        prefix_inline: Prefix for inline data ('I:')
        prefix_short: Prefix for short-term storage ('S:')
        prefix_persistent: Prefix for persistent storage ('P:')

    Example:
        >>> encoder = CallbackEncoder(storage)
        >>> action = MenuAction(handler="test", params={"id": 123})
        >>> encoded = await encoder.encode(action)
        >>> decoded = await encoder.decode(encoded)
    """

    # Constants for size thresholds (in bytes)
    INLINE_THRESHOLD = 60  # Telegram callback_data is 64 bytes, leave 4 for prefix
    SHORT_THRESHOLD = 500

    # Prefixes for different storage strategies
    PREFIX_INLINE = "I:"
    PREFIX_INLINE_COMPRESSED = "IC:"
    PREFIX_SHORT = "S:"
    PREFIX_PERSISTENT = "P:"

    def __init__(self, storage: StorageBackend) -> None:
        """Initialize encoder with storage backend.

        Args:
            storage: Storage backend for non-inline data
        """
        self.storage = storage

    async def encode(
        self, action: MenuAction, force_strategy: StorageStrategy | None = None
    ) -> str:
        """Encode MenuAction into callback_data string.

        Args:
            action: MenuAction to encode
            force_strategy: Force a specific storage strategy (for testing)

        Returns:
            Encoded callback_data string (max 64 bytes)

        Raises:
            EncodingError: If encoding fails
        """
        try:
            # Prepare data structure
            data = {
                "h": action.handler,  # Use short keys to save space
                "p": action.params,
            }

            # Try inline encoding first (unless forced otherwise)
            if force_strategy is None or force_strategy == StorageStrategy.INLINE:
                inline_encoded = self._encode_inline(data)
                if inline_encoded and len(inline_encoded) <= 64:
                    return inline_encoded

            # Determine storage strategy
            json_size = len(json.dumps(data, separators=(",", ":")))

            if force_strategy:
                strategy = force_strategy
            elif json_size < self.SHORT_THRESHOLD:
                strategy = StorageStrategy.SHORT
            else:
                strategy = StorageStrategy.PERSISTENT

            # Store externally and return reference
            key = self._generate_key(data)

            if strategy == StorageStrategy.SHORT:
                await self.storage.set(key, data, ttl=action.ttl)
                return f"{self.PREFIX_SHORT}{key}"
            # PERSISTENT
            await self.storage.set(key, data, ttl=None)
            return f"{self.PREFIX_PERSISTENT}{key}"

        except Exception as e:
            raise EncodingError(f"Failed to encode callback data: {e}") from e

    def encode_inline(self, action: MenuAction) -> str:
        """Encode a MenuAction using only the inline (storage-free) path.

        This is the synchronous counterpart to :meth:`encode`. It builds the same
        ``{"h": handler, "p": params}`` dict and runs *only* the inline strategy
        (JSON -> zlib -> base64). Unlike :meth:`encode`, it never spills to storage
        and performs no ``await``: if the resulting callback_data would not fit
        inline (i.e. it would otherwise require a storage spill), an
        :class:`EncodingError` is raised instead.

        Args:
            action: MenuAction to encode.

        Returns:
            Encoded callback_data string (``I:``/``IC:`` prefix, max 64 bytes).

        Raises:
            EncodingError: If the action does not fit within the 64-byte inline
                budget and would therefore require storage.
        """
        data = {
            "h": action.handler,  # Use short keys to save space
            "p": action.params,
        }

        inline_encoded = self._encode_inline(data)
        if inline_encoded is not None and len(inline_encoded) <= 64:
            return inline_encoded

        json_size = len(json.dumps(data, separators=(",", ":")).encode("utf-8"))
        message = (
            f"Callback for handler {action.handler!r} is {json_size}B encoded and exceeds the "
            "64B inline budget; it would require storage — use build_async() or shrink the params."
        )
        raise EncodingError(message)

    async def decode(self, callback_data: str) -> MenuAction:
        """Decode callback_data string back to MenuAction.

        Args:
            callback_data: Encoded callback_data string

        Returns:
            Decoded MenuAction

        Raises:
            DecodingError: If decoding fails or data not found
        """
        try:
            # Check prefix to determine decoding strategy
            if callback_data.startswith((self.PREFIX_INLINE_COMPRESSED, self.PREFIX_INLINE)):
                data = self._decode_inline(callback_data)
            elif callback_data.startswith(self.PREFIX_SHORT):
                key = callback_data[len(self.PREFIX_SHORT) :]
                stored_data = await self.storage.get(key)
                if stored_data is None:
                    raise DecodingError(f"Callback data expired or not found: {key}")
                data = stored_data
            elif callback_data.startswith(self.PREFIX_PERSISTENT):
                key = callback_data[len(self.PREFIX_PERSISTENT) :]
                stored_data = await self.storage.get(key)
                if stored_data is None:
                    raise DecodingError(f"Callback data not found: {key}")
                data = stored_data
            else:
                raise DecodingError(f"Unknown callback_data format: {callback_data[:10]}...")

            # Reconstruct MenuAction
            return MenuAction(
                handler=data["h"],
                params=data.get("p", {}),
            )

        except Exception as e:
            if isinstance(e, DecodingError):
                raise
            raise DecodingError(f"Failed to decode callback data: {e}") from e

    def _encode_inline(self, data: dict[str, Any]) -> str | None:
        """Encode data inline, preferring the most compact representation.

        Two inline tiers are considered and the smallest one that fits Telegram's
        64-byte ``callback_data`` limit wins:

        - ``I:`` carries the minified, ASCII-safe JSON verbatim (no base64
          overhead). This is the common case for small payloads.
        - ``IC:`` carries zlib-compressed, base64-encoded JSON. It is only used
          when compression actually produces a shorter result than the raw JSON
          (e.g. highly repetitive params).

        Args:
            data: Data dictionary to encode.

        Returns:
            Encoded string, or ``None`` if neither tier fits within 64 bytes.
        """
        try:
            # Serialize to JSON with minimal separators (ASCII-safe for callback_data).
            json_str = json.dumps(data, separators=(",", ":"), ensure_ascii=True)
            json_bytes = json_str.encode("utf-8")

            candidates: list[str] = []

            # Tier 1: raw JSON verbatim (no base64 overhead).
            raw = f"{self.PREFIX_INLINE}{json_str}"
            if len(raw.encode("utf-8")) <= 64:
                candidates.append(raw)

            # Tier 2: zlib + base64, only worthwhile when it beats the raw JSON.
            compressed = zlib.compress(json_bytes, level=9)
            if len(compressed) < len(json_bytes):
                b64_encoded = base64.b64encode(compressed).decode("ascii")
                comp = f"{self.PREFIX_INLINE_COMPRESSED}{b64_encoded}"
                if len(comp.encode("utf-8")) <= 64:
                    candidates.append(comp)

            if not candidates:
                return None
            return min(candidates, key=len)

        except Exception:
            return None

    def _decode_inline(self, encoded: str) -> dict[str, Any]:
        """Decode inline callback data.

        Args:
            encoded: Encoded string

        Returns:
            Decoded data dictionary

        Raises:
            DecodingError: If decoding fails
        """
        try:
            if encoded.startswith(self.PREFIX_INLINE_COMPRESSED):
                # Compressed tier: base64 -> zlib -> JSON.
                b64_str = encoded[len(self.PREFIX_INLINE_COMPRESSED) :]
                decoded_bytes = zlib.decompress(base64.b64decode(b64_str))
                json_str = decoded_bytes.decode("utf-8")
            elif encoded.startswith(self.PREFIX_INLINE):
                # Raw tier: the payload is the minified JSON verbatim.
                json_str = encoded[len(self.PREFIX_INLINE) :]
            else:
                raise DecodingError("Invalid inline encoding prefix")

            result: dict[str, Any] = json.loads(json_str)
            return result

        except Exception as e:
            raise DecodingError(f"Failed to decode inline data: {e}") from e

    def _generate_key(self, data: dict[str, Any]) -> str:
        """Generate deterministic key from data.

        This allows the same data to use the same storage key, enabling deduplication.

        Args:
            data: Data dictionary

        Returns:
            12-character hex hash
        """
        # Serialize with sorted keys for determinism. MD5 is used purely as a
        # fast, deterministic dedup key (not for security), so usedforsecurity is
        # disabled to make that intent explicit and satisfy security linters.
        json_str = json.dumps(data, sort_keys=True, separators=(",", ":"))
        hash_obj = hashlib.md5(json_str.encode("utf-8"), usedforsecurity=False)
        return hash_obj.hexdigest()[:12]

    async def cleanup_callback(self, callback_data: str) -> bool:
        """Cleanup storage for a callback_data reference.

        This should be called after a callback is processed to free storage.
        Only affects SHORT strategy (persistent data is kept).

        Args:
            callback_data: Callback data string

        Returns:
            True if cleaned up, False otherwise
        """
        try:
            if callback_data.startswith(self.PREFIX_SHORT):
                key = callback_data[len(self.PREFIX_SHORT) :]
                return await self.storage.delete(key)
            return False
        except Exception:
            return False

encode async 

encode(action: MenuAction, force_strategy: StorageStrategy | None = None) -> str

Encode MenuAction into callback_data string.

Parameters:

Name Type Description Default
action MenuAction

MenuAction to encode

 required
force_strategy StorageStrategy | None

Force a specific storage strategy (for testing)

 None

Returns:

Type Description
str

Encoded callback_data string (max 64 bytes)

Raises:

Type Description
EncodingError

If encoding fails
Source code in src/telegram_menu_builder/encoding.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
async def encode(
    self, action: MenuAction, force_strategy: StorageStrategy | None = None
) -> str:
    """Encode MenuAction into callback_data string.

    Args:
        action: MenuAction to encode
        force_strategy: Force a specific storage strategy (for testing)

    Returns:
        Encoded callback_data string (max 64 bytes)

    Raises:
        EncodingError: If encoding fails
    """
    try:
        # Prepare data structure
        data = {
            "h": action.handler,  # Use short keys to save space
            "p": action.params,
        }

        # Try inline encoding first (unless forced otherwise)
        if force_strategy is None or force_strategy == StorageStrategy.INLINE:
            inline_encoded = self._encode_inline(data)
            if inline_encoded and len(inline_encoded) <= 64:
                return inline_encoded

        # Determine storage strategy
        json_size = len(json.dumps(data, separators=(",", ":")))

        if force_strategy:
            strategy = force_strategy
        elif json_size < self.SHORT_THRESHOLD:
            strategy = StorageStrategy.SHORT
        else:
            strategy = StorageStrategy.PERSISTENT

        # Store externally and return reference
        key = self._generate_key(data)

        if strategy == StorageStrategy.SHORT:
            await self.storage.set(key, data, ttl=action.ttl)
            return f"{self.PREFIX_SHORT}{key}"
        # PERSISTENT
        await self.storage.set(key, data, ttl=None)
        return f"{self.PREFIX_PERSISTENT}{key}"

    except Exception as e:
        raise EncodingError(f"Failed to encode callback data: {e}") from e

encode_inline 

encode_inline(action: MenuAction) -> str

Encode a MenuAction using only the inline (storage-free) path.

This is the synchronous counterpart to :meth:encode. It builds the same {"h": handler, "p": params} dict and runs only the inline strategy (JSON -> zlib -> base64). Unlike :meth:encode, it never spills to storage and performs no await: if the resulting callback_data would not fit inline (i.e. it would otherwise require a storage spill), an :class:EncodingError is raised instead.

Parameters:

Name Type Description Default
action MenuAction

MenuAction to encode.

 required

Returns:

Type Description
str

Encoded callback_data string (I:/IC: prefix, max 64 bytes).

Raises:

Type Description
EncodingError

If the action does not fit within the 64-byte inline budget and would therefore require storage.
Source code in src/telegram_menu_builder/encoding.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def encode_inline(self, action: MenuAction) -> str:
    """Encode a MenuAction using only the inline (storage-free) path.

    This is the synchronous counterpart to :meth:`encode`. It builds the same
    ``{"h": handler, "p": params}`` dict and runs *only* the inline strategy
    (JSON -> zlib -> base64). Unlike :meth:`encode`, it never spills to storage
    and performs no ``await``: if the resulting callback_data would not fit
    inline (i.e. it would otherwise require a storage spill), an
    :class:`EncodingError` is raised instead.

    Args:
        action: MenuAction to encode.

    Returns:
        Encoded callback_data string (``I:``/``IC:`` prefix, max 64 bytes).

    Raises:
        EncodingError: If the action does not fit within the 64-byte inline
            budget and would therefore require storage.
    """
    data = {
        "h": action.handler,  # Use short keys to save space
        "p": action.params,
    }

    inline_encoded = self._encode_inline(data)
    if inline_encoded is not None and len(inline_encoded) <= 64:
        return inline_encoded

    json_size = len(json.dumps(data, separators=(",", ":")).encode("utf-8"))
    message = (
        f"Callback for handler {action.handler!r} is {json_size}B encoded and exceeds the "
        "64B inline budget; it would require storage — use build_async() or shrink the params."
    )
    raise EncodingError(message)

decode async 

decode(callback_data: str) -> MenuAction

Decode callback_data string back to MenuAction.

Parameters:

Name Type Description Default
callback_data str

Encoded callback_data string

 required

Returns:

Type Description
MenuAction

Decoded MenuAction

Raises:

Type Description
DecodingError

If decoding fails or data not found
Source code in src/telegram_menu_builder/encoding.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
async def decode(self, callback_data: str) -> MenuAction:
    """Decode callback_data string back to MenuAction.

    Args:
        callback_data: Encoded callback_data string

    Returns:
        Decoded MenuAction

    Raises:
        DecodingError: If decoding fails or data not found
    """
    try:
        # Check prefix to determine decoding strategy
        if callback_data.startswith((self.PREFIX_INLINE_COMPRESSED, self.PREFIX_INLINE)):
            data = self._decode_inline(callback_data)
        elif callback_data.startswith(self.PREFIX_SHORT):
            key = callback_data[len(self.PREFIX_SHORT) :]
            stored_data = await self.storage.get(key)
            if stored_data is None:
                raise DecodingError(f"Callback data expired or not found: {key}")
            data = stored_data
        elif callback_data.startswith(self.PREFIX_PERSISTENT):
            key = callback_data[len(self.PREFIX_PERSISTENT) :]
            stored_data = await self.storage.get(key)
            if stored_data is None:
                raise DecodingError(f"Callback data not found: {key}")
            data = stored_data
        else:
            raise DecodingError(f"Unknown callback_data format: {callback_data[:10]}...")

        # Reconstruct MenuAction
        return MenuAction(
            handler=data["h"],
            params=data.get("p", {}),
        )

    except Exception as e:
        if isinstance(e, DecodingError):
            raise
        raise DecodingError(f"Failed to decode callback data: {e}") from e

cleanup_callback async 

cleanup_callback(callback_data: str) -> bool

Cleanup storage for a callback_data reference.

This should be called after a callback is processed to free storage. Only affects SHORT strategy (persistent data is kept).

Parameters:

Name Type Description Default
callback_data str

Callback data string

 required

Returns:

Type Description
bool

True if cleaned up, False otherwise
Source code in src/telegram_menu_builder/encoding.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
async def cleanup_callback(self, callback_data: str) -> bool:
    """Cleanup storage for a callback_data reference.

    This should be called after a callback is processed to free storage.
    Only affects SHORT strategy (persistent data is kept).

    Args:
        callback_data: Callback data string

    Returns:
        True if cleaned up, False otherwise
    """
    try:
        if callback_data.startswith(self.PREFIX_SHORT):
            key = callback_data[len(self.PREFIX_SHORT) :]
            return await self.storage.delete(key)
        return False
    except Exception:
        return False

estimate_encoded_size

telegram_menu_builder.encoding.estimate_encoded_size 

estimate_encoded_size(action: MenuAction) -> int

Estimate the size of encoded callback data.

This is useful for determining storage strategy before encoding.

Parameters:

Name Type Description Default
action MenuAction

MenuAction to estimate

 required

Returns:

Type Description
int

Estimated size in bytes
Example

action = MenuAction(handler="test", params={"id": 123}) size = estimate_encoded_size(action) print(f"Estimated size: {size} bytes")

Source code in src/telegram_menu_builder/encoding.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def estimate_encoded_size(action: MenuAction) -> int:
    """Estimate the size of encoded callback data.

    This is useful for determining storage strategy before encoding.

    Args:
        action: MenuAction to estimate

    Returns:
        Estimated size in bytes

    Example:
        >>> action = MenuAction(handler="test", params={"id": 123})
        >>> size = estimate_encoded_size(action)
        >>> print(f"Estimated size: {size} bytes")
    """
    data = {
        "h": action.handler,
        "p": action.params,
    }

    json_str = json.dumps(data, separators=(",", ":"))
    json_bytes = json_str.encode("utf-8")

    # Estimate compressed size (rough approximation)
    # Real compression ratio varies, but this gives a reasonable estimate
    estimated_compressed = len(json_bytes) * 0.7  # Assume 30% compression

    # Base64 adds ~33% overhead
    estimated_b64 = estimated_compressed * 1.33

    # Add prefix (2-3 bytes)
    return int(estimated_b64 + 3)