Skip to content

Storage

Pluggable storage backends for callback data. Implement the StorageBackend protocol (or extend BaseStorage) to add your own.

StorageBackend

telegram_menu_builder.storage.base.StorageBackend

Bases: Protocol

Protocol defining the interface for storage backends.

All storage implementations must provide these methods to be compatible with the menu builder system.

Example

class MyStorage: ... async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None: ... # Implementation ... pass ... ... async def get(self, key: str) -> dict[str, Any] | None: ... # Implementation ... pass ... ... async def delete(self, key: str) -> bool: ... # Implementation ... pass

Source code in src/telegram_menu_builder/storage/base.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 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
@runtime_checkable
class StorageBackend(Protocol):
    """Protocol defining the interface for storage backends.

    All storage implementations must provide these methods to be compatible
    with the menu builder system.

    Example:
        >>> class MyStorage:
        ...     async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
        ...         # Implementation
        ...         pass
        ...
        ...     async def get(self, key: str) -> dict[str, Any] | None:
        ...         # Implementation
        ...         pass
        ...
        ...     async def delete(self, key: str) -> bool:
        ...         # Implementation
        ...         pass
    """

    async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
        """Store data with an optional TTL.

        Args:
            key: Unique identifier for the data
            data: Dictionary to store (must be JSON-serializable)
            ttl: Time-to-live in seconds (None = no expiration)

        Raises:
            StorageError: If storage operation fails
        """
        ...

    async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
        """Atomically store data only if the key is absent (set-if-absent).

        This is the single-winner primitive behind
        :meth:`~telegram_menu_builder.router.MenuRouter.claim`: a concurrent set
        of callers racing on the same key must see exactly one ``True`` return.
        An expired (non-live) value counts as absent and may be reclaimed.

        Args:
            key: Unique identifier for the data
            data: Dictionary to store (must be JSON-serializable)
            ttl: Time-to-live in seconds (None = no expiration)

        Returns:
            True if this call stored the value, False if a live (non-expired)
            value already existed for the key.

        Raises:
            StorageError: If storage operation fails
        """
        ...

    async def get(self, key: str) -> dict[str, Any] | None:
        """Retrieve data by key.

        Args:
            key: Unique identifier for the data

        Returns:
            Stored dictionary or None if not found/expired

        Raises:
            StorageError: If retrieval operation fails
        """
        ...

    async def delete(self, key: str) -> bool:
        """Delete data by key.

        Args:
            key: Unique identifier for the data

        Returns:
            True if deleted, False if not found

        Raises:
            StorageError: If deletion operation fails
        """
        ...

    async def exists(self, key: str) -> bool:
        """Check if key exists in storage.

        Args:
            key: Unique identifier to check

        Returns:
            True if key exists and hasn't expired
        """
        ...

    async def clear(self) -> None:
        """Clear all stored data.

        Warning:
            This will delete ALL data in the storage backend.
            Use with caution in production environments.

        Raises:
            StorageError: If clear operation fails
        """
        ...

    async def keys(self, pattern: str | None = None) -> list[str]:
        """Get all keys, optionally filtered by pattern.

        Args:
            pattern: Optional pattern for filtering keys (implementation-specific)

        Returns:
            List of matching keys

        Raises:
            StorageError: If operation fails
        """
        ...

set async 

set(key: str, data: dict[str, Any], ttl: int | None = None) -> None

Store data with an optional TTL.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required
data dict[str, Any]

Dictionary to store (must be JSON-serializable)

 required
ttl int | None

Time-to-live in seconds (None = no expiration)

 None

Raises:

Type Description
StorageError

If storage operation fails
Source code in src/telegram_menu_builder/storage/base.py 
33
34
35
36
37
38
39
40
41
42
43
44
async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
    """Store data with an optional TTL.

    Args:
        key: Unique identifier for the data
        data: Dictionary to store (must be JSON-serializable)
        ttl: Time-to-live in seconds (None = no expiration)

    Raises:
        StorageError: If storage operation fails
    """
    ...

add async 

add(key: str, data: dict[str, Any], ttl: int | None = None) -> bool

Atomically store data only if the key is absent (set-if-absent).

This is the single-winner primitive behind :meth:~telegram_menu_builder.router.MenuRouter.claim: a concurrent set of callers racing on the same key must see exactly one True return. An expired (non-live) value counts as absent and may be reclaimed.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required
data dict[str, Any]

Dictionary to store (must be JSON-serializable)

 required
ttl int | None

Time-to-live in seconds (None = no expiration)

 None

Returns:

Type Description
bool

True if this call stored the value, False if a live (non-expired)
bool

value already existed for the key.

Raises:

Type Description
StorageError

If storage operation fails
Source code in src/telegram_menu_builder/storage/base.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
    """Atomically store data only if the key is absent (set-if-absent).

    This is the single-winner primitive behind
    :meth:`~telegram_menu_builder.router.MenuRouter.claim`: a concurrent set
    of callers racing on the same key must see exactly one ``True`` return.
    An expired (non-live) value counts as absent and may be reclaimed.

    Args:
        key: Unique identifier for the data
        data: Dictionary to store (must be JSON-serializable)
        ttl: Time-to-live in seconds (None = no expiration)

    Returns:
        True if this call stored the value, False if a live (non-expired)
        value already existed for the key.

    Raises:
        StorageError: If storage operation fails
    """
    ...

get async 

get(key: str) -> dict[str, Any] | None

Retrieve data by key.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required

Returns:

Type Description
dict[str, Any] | None

Stored dictionary or None if not found/expired

Raises:

Type Description
StorageError

If retrieval operation fails
Source code in src/telegram_menu_builder/storage/base.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
async def get(self, key: str) -> dict[str, Any] | None:
    """Retrieve data by key.

    Args:
        key: Unique identifier for the data

    Returns:
        Stored dictionary or None if not found/expired

    Raises:
        StorageError: If retrieval operation fails
    """
    ...

delete async 

delete(key: str) -> bool

Delete data by key.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required

Returns:

Type Description
bool

True if deleted, False if not found

Raises:

Type Description
StorageError

If deletion operation fails
Source code in src/telegram_menu_builder/storage/base.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
async def delete(self, key: str) -> bool:
    """Delete data by key.

    Args:
        key: Unique identifier for the data

    Returns:
        True if deleted, False if not found

    Raises:
        StorageError: If deletion operation fails
    """
    ...

exists async 

exists(key: str) -> bool

Check if key exists in storage.

Parameters:

Name Type Description Default
key str

Unique identifier to check

 required

Returns:

Type Description
bool

True if key exists and hasn't expired
Source code in src/telegram_menu_builder/storage/base.py 
 96
 97
 98
 99
100
101
102
103
104
105
async def exists(self, key: str) -> bool:
    """Check if key exists in storage.

    Args:
        key: Unique identifier to check

    Returns:
        True if key exists and hasn't expired
    """
    ...

clear async 

clear() -> None

Clear all stored data.

Warning

This will delete ALL data in the storage backend. Use with caution in production environments.

Raises:

Type Description
StorageError

If clear operation fails
Source code in src/telegram_menu_builder/storage/base.py 
107
108
109
110
111
112
113
114
115
116
117
async def clear(self) -> None:
    """Clear all stored data.

    Warning:
        This will delete ALL data in the storage backend.
        Use with caution in production environments.

    Raises:
        StorageError: If clear operation fails
    """
    ...

keys async 

keys(pattern: str | None = None) -> list[str]

Get all keys, optionally filtered by pattern.

Parameters:

Name Type Description Default
pattern str | None

Optional pattern for filtering keys (implementation-specific)

 None

Returns:

Type Description
list[str]

List of matching keys

Raises:

Type Description
StorageError

If operation fails
Source code in src/telegram_menu_builder/storage/base.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
async def keys(self, pattern: str | None = None) -> list[str]:
    """Get all keys, optionally filtered by pattern.

    Args:
        pattern: Optional pattern for filtering keys (implementation-specific)

    Returns:
        List of matching keys

    Raises:
        StorageError: If operation fails
    """
    ...

BaseStorage

telegram_menu_builder.storage.base.BaseStorage

Bases: ABC

Abstract base class for storage backends with common functionality.

This class provides a template and shared utilities for storage implementations. Subclasses only need to implement the abstract methods.

Source code in src/telegram_menu_builder/storage/base.py 
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
class BaseStorage(ABC):
    """Abstract base class for storage backends with common functionality.

    This class provides a template and shared utilities for storage implementations.
    Subclasses only need to implement the abstract methods.
    """

    def __init__(self) -> None:
        """Initialize storage backend."""
        self._closed = False

    @abstractmethod
    async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
        """Store data with an optional TTL."""

    async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
        """Store data only if the key is absent (default, NON-ATOMIC implementation).

        This default ``exists``-then-``set`` implementation is provided so that
        custom backends written before ``add`` existed keep working without
        changes. It is a convenience, not a concurrency primitive.

        Warning:
            This default is NOT atomic: it awaits :meth:`exists` and then
            :meth:`set`, and another coroutine (or process) can interleave
            between the two. Concurrency-safe backends MUST override ``add`` with
            a genuinely atomic set-if-absent (e.g. Redis ``SET NX`` or a
            single-transaction conditional INSERT) so that racing callers see
            exactly one ``True``.

        Args:
            key: Unique identifier for the data
            data: Dictionary to store (must be JSON-serializable)
            ttl: Time-to-live in seconds (None = no expiration)

        Returns:
            True if this call stored the value, False if a live (non-expired)
            value already existed for the key.

        Raises:
            StorageError: If storage operation fails
        """
        if await self.exists(key):
            return False
        await self.set(key, data, ttl)
        return True

    @abstractmethod
    async def get(self, key: str) -> dict[str, Any] | None:
        """Retrieve data by key."""

    @abstractmethod
    async def delete(self, key: str) -> bool:
        """Delete data by key."""

    @abstractmethod
    async def exists(self, key: str) -> bool:
        """Check if key exists."""

    @abstractmethod
    async def clear(self) -> None:
        """Clear all stored data."""

    @abstractmethod
    async def keys(self, pattern: str | None = None) -> list[str]:
        """Get all keys, optionally filtered by pattern."""

    async def close(self) -> None:
        """Close storage backend and cleanup resources.

        Override this method if your storage needs cleanup (connections, files, etc.).
        """
        self._closed = True

    @property
    def is_closed(self) -> bool:
        """Check if storage is closed."""
        return self._closed

    def _ensure_open(self) -> None:
        """Ensure storage is not closed.

        Raises:
            RuntimeError: If storage is closed
        """
        if self._closed:
            raise RuntimeError(f"{self.__class__.__name__} is closed")

    async def __aenter__(self) -> "BaseStorage":
        """Async context manager entry."""
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """Async context manager exit."""
        await self.close()

is_closed property 

is_closed: bool

Check if storage is closed.

set abstractmethod async 

set(key: str, data: dict[str, Any], ttl: int | None = None) -> None

Store data with an optional TTL.

Source code in src/telegram_menu_builder/storage/base.py 
145
146
147
@abstractmethod
async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
    """Store data with an optional TTL."""

add async 

add(key: str, data: dict[str, Any], ttl: int | None = None) -> bool

Store data only if the key is absent (default, NON-ATOMIC implementation).

This default exists-then-set implementation is provided so that custom backends written before add existed keep working without changes. It is a convenience, not a concurrency primitive.

Warning

This default is NOT atomic: it awaits :meth:exists and then :meth:set, and another coroutine (or process) can interleave between the two. Concurrency-safe backends MUST override add with a genuinely atomic set-if-absent (e.g. Redis SET NX or a single-transaction conditional INSERT) so that racing callers see exactly one True.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required
data dict[str, Any]

Dictionary to store (must be JSON-serializable)

 required
ttl int | None

Time-to-live in seconds (None = no expiration)

 None

Returns:

Type Description
bool

True if this call stored the value, False if a live (non-expired)
bool

value already existed for the key.

Raises:

Type Description
StorageError

If storage operation fails
Source code in src/telegram_menu_builder/storage/base.py 
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
async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
    """Store data only if the key is absent (default, NON-ATOMIC implementation).

    This default ``exists``-then-``set`` implementation is provided so that
    custom backends written before ``add`` existed keep working without
    changes. It is a convenience, not a concurrency primitive.

    Warning:
        This default is NOT atomic: it awaits :meth:`exists` and then
        :meth:`set`, and another coroutine (or process) can interleave
        between the two. Concurrency-safe backends MUST override ``add`` with
        a genuinely atomic set-if-absent (e.g. Redis ``SET NX`` or a
        single-transaction conditional INSERT) so that racing callers see
        exactly one ``True``.

    Args:
        key: Unique identifier for the data
        data: Dictionary to store (must be JSON-serializable)
        ttl: Time-to-live in seconds (None = no expiration)

    Returns:
        True if this call stored the value, False if a live (non-expired)
        value already existed for the key.

    Raises:
        StorageError: If storage operation fails
    """
    if await self.exists(key):
        return False
    await self.set(key, data, ttl)
    return True

get abstractmethod async 

get(key: str) -> dict[str, Any] | None

Retrieve data by key.

Source code in src/telegram_menu_builder/storage/base.py 
181
182
183
@abstractmethod
async def get(self, key: str) -> dict[str, Any] | None:
    """Retrieve data by key."""

delete abstractmethod async 

delete(key: str) -> bool

Delete data by key.

Source code in src/telegram_menu_builder/storage/base.py 
185
186
187
@abstractmethod
async def delete(self, key: str) -> bool:
    """Delete data by key."""

exists abstractmethod async 

exists(key: str) -> bool

Check if key exists.

Source code in src/telegram_menu_builder/storage/base.py 
189
190
191
@abstractmethod
async def exists(self, key: str) -> bool:
    """Check if key exists."""

clear abstractmethod async 

clear() -> None

Clear all stored data.

Source code in src/telegram_menu_builder/storage/base.py 
193
194
195
@abstractmethod
async def clear(self) -> None:
    """Clear all stored data."""

keys abstractmethod async 

keys(pattern: str | None = None) -> list[str]

Get all keys, optionally filtered by pattern.

Source code in src/telegram_menu_builder/storage/base.py 
197
198
199
@abstractmethod
async def keys(self, pattern: str | None = None) -> list[str]:
    """Get all keys, optionally filtered by pattern."""

close async 

close() -> None

Close storage backend and cleanup resources.

Override this method if your storage needs cleanup (connections, files, etc.).

Source code in src/telegram_menu_builder/storage/base.py 
201
202
203
204
205
206
async def close(self) -> None:
    """Close storage backend and cleanup resources.

    Override this method if your storage needs cleanup (connections, files, etc.).
    """
    self._closed = True

MemoryStorage

telegram_menu_builder.storage.memory.MemoryStorage

Bases: BaseStorage

In-memory storage backend using Python dictionaries.

This storage keeps all data in memory and supports TTL-based expiration. Data is lost when the process restarts.

Attributes:

Name Type Description
_data dict[str, dict[str, Any]]

Internal dictionary storing the data
_expiry dict[str, float]

Dictionary mapping keys to expiration timestamps
Example

storage = MemoryStorage() await storage.set("key1", {"value": 123}, ttl=60) data = await storage.get("key1") print(data) {'value': 123}

Thread Safety

This implementation is NOT thread-safe. If you need concurrent access, consider using locks or a thread-safe storage backend.

Source code in src/telegram_menu_builder/storage/memory.py 
 14
 15
 16
 17
 18
 19
 20
 21
 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
class MemoryStorage(BaseStorage):
    """In-memory storage backend using Python dictionaries.

    This storage keeps all data in memory and supports TTL-based expiration.
    Data is lost when the process restarts.

    Attributes:
        _data: Internal dictionary storing the data
        _expiry: Dictionary mapping keys to expiration timestamps

    Example:
        >>> storage = MemoryStorage()
        >>> await storage.set("key1", {"value": 123}, ttl=60)
        >>> data = await storage.get("key1")
        >>> print(data)
        {'value': 123}

    Thread Safety:
        This implementation is NOT thread-safe. If you need concurrent access,
        consider using locks or a thread-safe storage backend.
    """

    def __init__(self) -> None:
        """Initialize in-memory storage."""
        super().__init__()
        self._data: dict[str, dict[str, Any]] = {}
        self._expiry: dict[str, float] = {}

    async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
        """Store data in memory with optional TTL.

        Args:
            key: Unique identifier for the data
            data: Dictionary to store
            ttl: Time-to-live in seconds (None = no expiration)

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        self._data[key] = data.copy()  # Store a copy to prevent external modifications

        if ttl is not None:
            self._expiry[key] = time.time() + ttl
        elif key in self._expiry:
            # Remove expiry if it existed before
            del self._expiry[key]

    async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
        """Atomically store data only if the key is absent (set-if-absent).

        A key whose TTL has already elapsed is treated as absent and may be
        reclaimed. The existence check and the store happen with NO ``await``
        between them, so under a single event loop this is effectively atomic:
        two coroutines racing on the same key yield exactly one ``True``.

        Args:
            key: Unique identifier for the data
            data: Dictionary to store
            ttl: Time-to-live in seconds (None = no expiration)

        Returns:
            True if this call stored the value, False if a live (non-expired)
            value already existed for the key.

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        # A live (present and not-yet-expired) value blocks the claim.
        if key in self._data and not (key in self._expiry and time.time() > self._expiry[key]):
            return False

        # Absent or expired: claim it. No await between the check and the store,
        # so a concurrent add() cannot interleave under a single event loop.
        self._data[key] = data.copy()
        if ttl is not None:
            self._expiry[key] = time.time() + ttl
        elif key in self._expiry:
            del self._expiry[key]
        return True

    async def get(self, key: str) -> dict[str, Any] | None:
        """Retrieve data from memory.

        Args:
            key: Unique identifier for the data

        Returns:
            Stored dictionary or None if not found/expired

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        # Check if key exists
        if key not in self._data:
            return None

        # Check if expired
        if key in self._expiry and time.time() > self._expiry[key]:
            # Expired - clean up and return None
            await self.delete(key)
            return None

        # Return a copy to prevent external modifications
        return self._data[key].copy()

    async def delete(self, key: str) -> bool:
        """Delete data from memory.

        Args:
            key: Unique identifier for the data

        Returns:
            True if deleted, False if not found

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        existed = key in self._data

        if key in self._data:
            del self._data[key]

        if key in self._expiry:
            del self._expiry[key]

        return existed

    async def exists(self, key: str) -> bool:
        """Check if key exists in memory.

        Args:
            key: Unique identifier to check

        Returns:
            True if key exists and hasn't expired

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        if key not in self._data:
            return False

        # Check expiration
        if key in self._expiry and time.time() > self._expiry[key]:
            # Expired - clean up
            await self.delete(key)
            return False

        return True

    async def clear(self) -> None:
        """Clear all data from memory.

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        self._data.clear()
        self._expiry.clear()

    async def keys(self, pattern: str | None = None) -> list[str]:
        """Get all keys from memory, optionally filtered by pattern.

        Args:
            pattern: Optional glob-style pattern (supports * and ?)

        Returns:
            List of matching keys (excluding expired ones)

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        # Clean up expired keys first
        current_time = time.time()
        expired_keys = [key for key, expiry in self._expiry.items() if current_time > expiry]

        for key in expired_keys:
            await self.delete(key)

        # Get all valid keys
        all_keys = list(self._data.keys())

        # Filter by pattern if provided
        if pattern is None:
            return all_keys

        # Simple glob pattern matching
        return [key for key in all_keys if fnmatch.fnmatch(key, pattern)]

    async def cleanup_expired(self) -> int:
        """Manually cleanup expired entries.

        This method is useful for periodic cleanup in long-running applications.

        Returns:
            Number of expired entries removed

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        current_time = time.time()
        expired_keys = [key for key, expiry in self._expiry.items() if current_time > expiry]

        for key in expired_keys:
            await self.delete(key)

        return len(expired_keys)

    def get_stats(self) -> dict[str, Any]:
        """Get storage statistics.

        Returns:
            Dictionary with storage stats (total keys, expired keys, etc.)

        Raises:
            RuntimeError: If storage is closed
        """
        self._ensure_open()

        current_time = time.time()
        expired_count = sum(1 for expiry in self._expiry.values() if current_time > expiry)

        return {
            "total_keys": len(self._data),
            "keys_with_ttl": len(self._expiry),
            "expired_keys": expired_count,
            "active_keys": len(self._data) - expired_count,
        }

    async def close(self) -> None:
        """Close storage and free memory.

        After closing, the storage cannot be used anymore.
        """
        if not self._closed:
            self._data.clear()
            self._expiry.clear()
            await super().close()

set async 

set(key: str, data: dict[str, Any], ttl: int | None = None) -> None

Store data in memory with optional TTL.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required
data dict[str, Any]

Dictionary to store

 required
ttl int | None

Time-to-live in seconds (None = no expiration)

 None

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
    """Store data in memory with optional TTL.

    Args:
        key: Unique identifier for the data
        data: Dictionary to store
        ttl: Time-to-live in seconds (None = no expiration)

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    self._data[key] = data.copy()  # Store a copy to prevent external modifications

    if ttl is not None:
        self._expiry[key] = time.time() + ttl
    elif key in self._expiry:
        # Remove expiry if it existed before
        del self._expiry[key]

add async 

add(key: str, data: dict[str, Any], ttl: int | None = None) -> bool

Atomically store data only if the key is absent (set-if-absent).

A key whose TTL has already elapsed is treated as absent and may be reclaimed. The existence check and the store happen with NO await between them, so under a single event loop this is effectively atomic: two coroutines racing on the same key yield exactly one True.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required
data dict[str, Any]

Dictionary to store

 required
ttl int | None

Time-to-live in seconds (None = no expiration)

 None

Returns:

Type Description
bool

True if this call stored the value, False if a live (non-expired)
bool

value already existed for the key.

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
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
async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
    """Atomically store data only if the key is absent (set-if-absent).

    A key whose TTL has already elapsed is treated as absent and may be
    reclaimed. The existence check and the store happen with NO ``await``
    between them, so under a single event loop this is effectively atomic:
    two coroutines racing on the same key yield exactly one ``True``.

    Args:
        key: Unique identifier for the data
        data: Dictionary to store
        ttl: Time-to-live in seconds (None = no expiration)

    Returns:
        True if this call stored the value, False if a live (non-expired)
        value already existed for the key.

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    # A live (present and not-yet-expired) value blocks the claim.
    if key in self._data and not (key in self._expiry and time.time() > self._expiry[key]):
        return False

    # Absent or expired: claim it. No await between the check and the store,
    # so a concurrent add() cannot interleave under a single event loop.
    self._data[key] = data.copy()
    if ttl is not None:
        self._expiry[key] = time.time() + ttl
    elif key in self._expiry:
        del self._expiry[key]
    return True

get async 

get(key: str) -> dict[str, Any] | None

Retrieve data from memory.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required

Returns:

Type Description
dict[str, Any] | None

Stored dictionary or None if not found/expired

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
 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
async def get(self, key: str) -> dict[str, Any] | None:
    """Retrieve data from memory.

    Args:
        key: Unique identifier for the data

    Returns:
        Stored dictionary or None if not found/expired

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    # Check if key exists
    if key not in self._data:
        return None

    # Check if expired
    if key in self._expiry and time.time() > self._expiry[key]:
        # Expired - clean up and return None
        await self.delete(key)
        return None

    # Return a copy to prevent external modifications
    return self._data[key].copy()

delete async 

delete(key: str) -> bool

Delete data from memory.

Parameters:

Name Type Description Default
key str

Unique identifier for the data

 required

Returns:

Type Description
bool

True if deleted, False if not found

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def delete(self, key: str) -> bool:
    """Delete data from memory.

    Args:
        key: Unique identifier for the data

    Returns:
        True if deleted, False if not found

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    existed = key in self._data

    if key in self._data:
        del self._data[key]

    if key in self._expiry:
        del self._expiry[key]

    return existed

exists async 

exists(key: str) -> bool

Check if key exists in memory.

Parameters:

Name Type Description Default
key str

Unique identifier to check

 required

Returns:

Type Description
bool

True if key exists and hasn't expired

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
async def exists(self, key: str) -> bool:
    """Check if key exists in memory.

    Args:
        key: Unique identifier to check

    Returns:
        True if key exists and hasn't expired

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    if key not in self._data:
        return False

    # Check expiration
    if key in self._expiry and time.time() > self._expiry[key]:
        # Expired - clean up
        await self.delete(key)
        return False

    return True

clear async 

clear() -> None

Clear all data from memory.

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
174
175
176
177
178
179
180
181
182
183
async def clear(self) -> None:
    """Clear all data from memory.

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    self._data.clear()
    self._expiry.clear()

keys async 

keys(pattern: str | None = None) -> list[str]

Get all keys from memory, optionally filtered by pattern.

Parameters:

Name Type Description Default
pattern str | None

Optional glob-style pattern (supports * and ?)

 None

Returns:

Type Description
list[str]

List of matching keys (excluding expired ones)

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
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
async def keys(self, pattern: str | None = None) -> list[str]:
    """Get all keys from memory, optionally filtered by pattern.

    Args:
        pattern: Optional glob-style pattern (supports * and ?)

    Returns:
        List of matching keys (excluding expired ones)

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    # Clean up expired keys first
    current_time = time.time()
    expired_keys = [key for key, expiry in self._expiry.items() if current_time > expiry]

    for key in expired_keys:
        await self.delete(key)

    # Get all valid keys
    all_keys = list(self._data.keys())

    # Filter by pattern if provided
    if pattern is None:
        return all_keys

    # Simple glob pattern matching
    return [key for key in all_keys if fnmatch.fnmatch(key, pattern)]

cleanup_expired async 

cleanup_expired() -> int

Manually cleanup expired entries.

This method is useful for periodic cleanup in long-running applications.

Returns:

Type Description
int

Number of expired entries removed

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
async def cleanup_expired(self) -> int:
    """Manually cleanup expired entries.

    This method is useful for periodic cleanup in long-running applications.

    Returns:
        Number of expired entries removed

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    current_time = time.time()
    expired_keys = [key for key, expiry in self._expiry.items() if current_time > expiry]

    for key in expired_keys:
        await self.delete(key)

    return len(expired_keys)

get_stats 

get_stats() -> dict[str, Any]

Get storage statistics.

Returns:

Type Description
dict[str, Any]

Dictionary with storage stats (total keys, expired keys, etc.)

Raises:

Type Description
RuntimeError

If storage is closed
Source code in src/telegram_menu_builder/storage/memory.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def get_stats(self) -> dict[str, Any]:
    """Get storage statistics.

    Returns:
        Dictionary with storage stats (total keys, expired keys, etc.)

    Raises:
        RuntimeError: If storage is closed
    """
    self._ensure_open()

    current_time = time.time()
    expired_count = sum(1 for expiry in self._expiry.values() if current_time > expiry)

    return {
        "total_keys": len(self._data),
        "keys_with_ttl": len(self._expiry),
        "expired_keys": expired_count,
        "active_keys": len(self._data) - expired_count,
    }

close async 

close() -> None

Close storage and free memory.

After closing, the storage cannot be used anymore.

Source code in src/telegram_menu_builder/storage/memory.py 
258
259
260
261
262
263
264
265
266
async def close(self) -> None:
    """Close storage and free memory.

    After closing, the storage cannot be used anymore.
    """
    if not self._closed:
        self._data.clear()
        self._expiry.clear()
        await super().close()

SQLAlchemyStorage

telegram_menu_builder.storage.sqlalchemy.SQLAlchemyStorage

Bases: BaseStorage

Async SQL storage backend using SQLAlchemy 2.0 Core.

Stores callback payloads in a single table with columns key (bounded string primary key), value (JSON) and expires_at (nullable, indexed timezone-aware datetime). Works across PostgreSQL/Supabase, MySQL/MariaDB and SQLite from one code path; only the UPSERT statement is dialect-aware.

Engine ownership

If constructed from a database_url the backend creates and OWNS the engine and disposes it on :meth:close. If an existing engine is supplied the backend BORROWS it and never disposes it.

Example

store = SQLAlchemyStorage(database_url="sqlite+aiosqlite:///:memory:") await store.create_schema() await store.set("k", {"h": "menu", "p": {"x": 1}}) await store.get("k") {'h': 'menu', 'p': {'x': 1}} await store.close()

Note

Unlike :class:~telegram_menu_builder.storage.memory.MemoryStorage, :meth:get_stats here is asynchronous because it queries the database.

Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
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
302
303
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
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
class SQLAlchemyStorage(BaseStorage):
    """Async SQL storage backend using SQLAlchemy 2.0 Core.

    Stores callback payloads in a single table with columns ``key`` (bounded
    string primary key), ``value`` (JSON) and ``expires_at`` (nullable, indexed
    timezone-aware datetime). Works across PostgreSQL/Supabase, MySQL/MariaDB
    and SQLite from one code path; only the UPSERT statement is dialect-aware.

    Engine ownership:
        If constructed from a ``database_url`` the backend creates and OWNS the
        engine and disposes it on :meth:`close`. If an existing ``engine`` is
        supplied the backend BORROWS it and never disposes it.

    Example:
        >>> store = SQLAlchemyStorage(database_url="sqlite+aiosqlite:///:memory:")
        >>> await store.create_schema()
        >>> await store.set("k", {"h": "menu", "p": {"x": 1}})
        >>> await store.get("k")
        {'h': 'menu', 'p': {'x': 1}}
        >>> await store.close()

    Note:
        Unlike :class:`~telegram_menu_builder.storage.memory.MemoryStorage`,
        :meth:`get_stats` here is asynchronous because it queries the database.
    """

    def __init__(
        self,
        database_url: str | None = None,
        *,
        engine: AsyncEngine | None = None,
        table_name: str = "menu_callbacks",
        schema: str | None = None,
    ) -> None:
        """Initialize the SQL storage backend.

        Exactly one of ``database_url`` or ``engine`` must be supplied.

        Args:
            database_url: An async SQLAlchemy URL (e.g.
                ``"postgresql+asyncpg://..."``, ``"mysql+asyncmy://..."`` or
                ``"sqlite+aiosqlite:///:memory:"``). When given, an engine is
                created and owned by this instance.
            engine: An existing :class:`~sqlalchemy.ext.asyncio.AsyncEngine` to
                borrow. When given, it is NOT disposed on :meth:`close`.
            table_name: Name of the table holding callback payloads.
            schema: Optional database schema/namespace for the table.

        Raises:
            ValueError: If neither or both of ``database_url`` and ``engine``
                are provided.
        """
        super().__init__()

        if (database_url is None) == (engine is None):
            msg = (
                "Exactly one of 'database_url' or 'engine' must be provided (got both or neither)."
            )
            raise ValueError(msg)

        self._owns_engine: bool
        self._engine: AsyncEngine
        if database_url is not None:
            self._engine = self._create_engine(database_url)
            self._owns_engine = True
        else:
            # engine is not None here (guaranteed by the XOR check above).
            assert engine is not None
            self._engine = engine
            self._owns_engine = False

        self._metadata = MetaData()
        self._table = Table(
            table_name,
            self._metadata,
            Column("key", String(255), primary_key=True),
            Column("value", JSON, nullable=False),
            Column("expires_at", UtcDateTime, nullable=True, index=True),
            schema=schema,
        )

    @staticmethod
    def _create_engine(database_url: str) -> AsyncEngine:
        """Create an owned async engine from a URL.

        In-memory SQLite URLs require a :class:`~sqlalchemy.pool.StaticPool` and
        ``check_same_thread=False`` so the single underlying connection persists
        across operations; otherwise the async pool would hand out fresh
        connections and lose the data.

        Args:
            database_url: The async database URL.

        Returns:
            A newly created :class:`~sqlalchemy.ext.asyncio.AsyncEngine`.
        """
        if ":memory:" in database_url:
            return create_async_engine(
                database_url,
                poolclass=StaticPool,
                connect_args={"check_same_thread": False},
            )
        return create_async_engine(database_url)

    def _not_expired(self, now: datetime.datetime) -> ColumnElement[bool]:
        """Build the "not expired" predicate for queries.

        Args:
            now: The reference moment (timezone-aware UTC).

        Returns:
            A boolean column expression that is true for rows without an expiry
            or whose expiry is strictly in the future.
        """
        expires_at = self._table.c.expires_at
        return (expires_at.is_(None)) | (expires_at > now)

    async def create_schema(self) -> None:
        """Create the backing table and indexes if they do not exist.

        This is idempotent: it uses ``checkfirst=True`` so it can be called
        repeatedly without error.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the DDL operation fails.
        """
        self._ensure_open()
        try:
            async with self._engine.begin() as conn:
                await conn.run_sync(self._metadata.create_all, checkfirst=True)
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to create schema: {exc}") from exc

    async def drop_schema(self) -> None:
        """Drop the backing table if it exists.

        This is idempotent thanks to ``checkfirst=True``.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the DDL operation fails.
        """
        self._ensure_open()
        try:
            async with self._engine.begin() as conn:
                await conn.run_sync(self._metadata.drop_all, checkfirst=True)
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to drop schema: {exc}") from exc

    async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
        """Store (UPSERT) data under ``key`` with an optional TTL.

        Re-setting the same key overwrites the existing value and refreshes (or
        clears) its expiry, matching the encoder's deterministic-key dedup.

        Args:
            key: Unique identifier for the data.
            data: JSON-serializable dictionary to store.
            ttl: Time-to-live in seconds (``None`` = no expiration).

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the write fails.
        """
        self._ensure_open()

        expires_at = _utcnow() + datetime.timedelta(seconds=ttl) if ttl is not None else None
        values = {"key": key, "value": data, "expires_at": expires_at}

        try:
            stmt = self._build_upsert(values)
            if stmt is None:
                # Dialect without native UPSERT: fall back to DELETE-then-INSERT.
                await self._fallback_set(values)
            else:
                async with self._engine.begin() as conn:
                    await conn.execute(stmt)
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to set key {key!r}: {exc}") from exc

    def _build_upsert(self, values: dict[str, Any]) -> Any:
        """Build a dialect-specific UPSERT statement.

        Branches on ``self._engine.dialect.name``:

        * ``postgresql`` / ``sqlite``: ``INSERT ... ON CONFLICT DO UPDATE``.
        * ``mysql`` / ``mariadb``: ``INSERT ... ON DUPLICATE KEY UPDATE``.
        * otherwise (fallback): a delete-by-key statement; :meth:`set` runs the
          fallback as DELETE-then-INSERT inside one transaction. Because the
          fallback needs two statements, ``set`` handles it specially when this
          method returns ``None``.

        Args:
            values: The row values to insert (``key``, ``value``, ``expires_at``).

        Returns:
            An executable insert statement for known dialects.
        """
        dialect = self._engine.dialect.name

        if dialect in ("postgresql", "sqlite"):
            pg_or_sqlite = pg_insert if dialect == "postgresql" else sqlite_insert
            stmt = pg_or_sqlite(self._table).values(**values)
            return stmt.on_conflict_do_update(
                index_elements=[self._table.c.key],
                set_={
                    "value": stmt.excluded.value,
                    "expires_at": stmt.excluded.expires_at,
                },
            )

        if dialect in ("mysql", "mariadb"):
            my_stmt = mysql_insert(self._table).values(**values)
            return my_stmt.on_duplicate_key_update(
                value=my_stmt.inserted.value,
                expires_at=my_stmt.inserted.expires_at,
            )

        # Fallback for any other dialect: signal DELETE-then-INSERT to set().
        return None

    async def _fallback_set(self, values: dict[str, Any]) -> None:
        """Emulate UPSERT on dialects without native conflict handling.

        Performs ``DELETE WHERE key`` then ``INSERT`` inside a single
        transaction so the operation is atomic.

        Args:
            values: The row values (``key``, ``value``, ``expires_at``).
        """
        async with self._engine.begin() as conn:
            await conn.execute(delete(self._table).where(self._table.c.key == values["key"]))
            await conn.execute(self._table.insert().values(**values))

    async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
        """Atomically store ``data`` under ``key`` only if it is absent.

        The whole operation runs inside a SINGLE transaction
        (``async with self._engine.begin()``) so it is atomic against concurrent
        callers and the engine's connection pool. Within that transaction it:

        1. DELETEs any row for ``key`` whose TTL has already elapsed
           (``expires_at IS NOT NULL AND expires_at <= now``), freeing an expired
           claim so it can be reclaimed; and then
        2. runs a dialect INSERT that no-ops on a primary-key conflict, mirroring
           the dialect branches of :meth:`_build_upsert`: PostgreSQL/SQLite use
           ``INSERT ... ON CONFLICT DO NOTHING``; MySQL/MariaDB use
           ``INSERT IGNORE``; any other dialect falls back to a plain INSERT and
           treats an :class:`~sqlalchemy.exc.IntegrityError` as a lost race.

        The boolean result is derived from the INSERT's affected-row count: a row
        was inserted (won the claim) iff ``rowcount > 0``.

        Args:
            key: Unique identifier for the data.
            data: JSON-serializable dictionary to store.
            ttl: Time-to-live in seconds (``None`` = no expiration).

        Returns:
            ``True`` if this call stored the value, ``False`` if a live
            (non-expired) row already existed for the key.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the write fails.
        """
        self._ensure_open()

        now = _utcnow()
        expires_at = now + datetime.timedelta(seconds=ttl) if ttl is not None else None
        values = {"key": key, "value": data, "expires_at": expires_at}

        try:
            async with self._engine.begin() as conn:
                # Free an expired claim first so it can be reclaimed in this txn.
                expires_col = self._table.c.expires_at
                await conn.execute(
                    delete(self._table).where(
                        (self._table.c.key == key)
                        & (expires_col.is_not(None))
                        & (expires_col <= now)
                    )
                )

                stmt = self._build_insert_ignore(values)
                if stmt is None:
                    # Dialect without a native no-op insert: try a plain INSERT and
                    # treat a primary-key collision as a lost race.
                    try:
                        result = await conn.execute(self._table.insert().values(**values))
                    except IntegrityError:
                        return False
                    return result.rowcount > 0

                result = await conn.execute(stmt)
                return result.rowcount > 0
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to add key {key!r}: {exc}") from exc

    def _build_insert_ignore(self, values: dict[str, Any]) -> Any:
        """Build a dialect-specific INSERT that no-ops on a primary-key conflict.

        Branches on ``self._engine.dialect.name``, mirroring :meth:`_build_upsert`:

        * ``postgresql`` / ``sqlite``: ``INSERT ... ON CONFLICT DO NOTHING``.
        * ``mysql`` / ``mariadb``: ``INSERT IGNORE``.
        * otherwise (fallback): ``None``, signalling :meth:`add` to attempt a
          plain INSERT and catch :class:`~sqlalchemy.exc.IntegrityError`.

        Args:
            values: The row values to insert (``key``, ``value``, ``expires_at``).

        Returns:
            An executable conflict-tolerant insert statement for known dialects,
            or ``None`` to request the generic INSERT/``IntegrityError`` fallback.
        """
        dialect = self._engine.dialect.name

        if dialect in ("postgresql", "sqlite"):
            pg_or_sqlite = pg_insert if dialect == "postgresql" else sqlite_insert
            stmt = pg_or_sqlite(self._table).values(**values)
            return stmt.on_conflict_do_nothing(index_elements=[self._table.c.key])

        if dialect in ("mysql", "mariadb"):
            return mysql_insert(self._table).values(**values).prefix_with("IGNORE")

        # Fallback for any other dialect: signal plain INSERT to add().
        return None

    async def get(self, key: str) -> dict[str, Any] | None:
        """Retrieve the (non-expired) value stored under ``key``.

        Expiry is filtered lazily at query time; an expired row is treated as
        missing and ``None`` is returned (the row is not eagerly deleted here).

        Args:
            key: Unique identifier for the data.

        Returns:
            The stored dictionary, or ``None`` if missing or expired.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the read fails.
        """
        self._ensure_open()

        now = _utcnow()
        stmt = select(self._table.c.value).where(
            (self._table.c.key == key) & self._not_expired(now)
        )
        try:
            async with self._engine.connect() as conn:
                result = await conn.execute(stmt)
                raw = result.scalar_one_or_none()
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to get key {key!r}: {exc}") from exc

        if raw is None:
            return None
        return cast("dict[str, Any]", raw)

    async def delete(self, key: str) -> bool:
        """Delete the row stored under ``key``.

        Args:
            key: Unique identifier for the data.

        Returns:
            ``True`` if a row was deleted, ``False`` if the key did not exist.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the delete fails.
        """
        self._ensure_open()

        stmt = delete(self._table).where(self._table.c.key == key)
        try:
            async with self._engine.begin() as conn:
                result = await conn.execute(stmt)
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to delete key {key!r}: {exc}") from exc

        return result.rowcount > 0

    async def exists(self, key: str) -> bool:
        """Check whether a non-expired row exists under ``key``.

        Args:
            key: Unique identifier to check.

        Returns:
            ``True`` if the key exists and has not expired.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the read fails.
        """
        self._ensure_open()

        now = _utcnow()
        stmt = select(self._table.c.key).where((self._table.c.key == key) & self._not_expired(now))
        try:
            async with self._engine.connect() as conn:
                result = await conn.execute(stmt)
                row = result.first()
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to check existence of key {key!r}: {exc}") from exc

        return row is not None

    async def clear(self) -> None:
        """Delete every row in the backing table.

        Warning:
            This removes ALL stored callbacks. Use with care in production.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the delete fails.
        """
        self._ensure_open()

        try:
            async with self._engine.begin() as conn:
                await conn.execute(delete(self._table))
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to clear storage: {exc}") from exc

    async def keys(self, pattern: str | None = None) -> list[str]:
        """Return all non-expired keys, optionally filtered by a glob pattern.

        The glob ``pattern`` is translated to a SQL ``LIKE`` clause and pushed
        down to the database.

        Note:
            This diverges from :class:`MemoryStorage`'s ``fnmatch`` semantics:
            SQL ``LIKE`` supports only ``*``/``?`` (mapped to ``%``/``_``) and
            has no ``[seq]`` character-class wildcards.

        Args:
            pattern: Optional glob pattern using ``*`` and ``?``.

        Returns:
            A list of matching, non-expired keys.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the read fails.
        """
        self._ensure_open()

        now = _utcnow()
        condition: ColumnElement[bool] = self._not_expired(now)
        if pattern is not None:
            like_clause: BinaryExpression[bool] = self._table.c.key.like(
                _glob_to_like(pattern), escape="\\"
            )
            condition = condition & like_clause

        stmt = select(self._table.c.key).where(condition)
        try:
            async with self._engine.connect() as conn:
                result = await conn.execute(stmt)
                rows = result.scalars().all()
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to list keys: {exc}") from exc

        return list(rows)

    async def cleanup_expired(self) -> int:
        """Delete all rows whose TTL has elapsed.

        Returns:
            The number of expired rows removed.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the delete fails.
        """
        self._ensure_open()

        now = _utcnow()
        expires_at = self._table.c.expires_at
        stmt = delete(self._table).where((expires_at.is_not(None)) & (expires_at <= now))
        try:
            async with self._engine.begin() as conn:
                result = await conn.execute(stmt)
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to clean up expired keys: {exc}") from exc

        return result.rowcount

    async def get_stats(self) -> dict[str, int]:
        """Return counts describing the stored keys.

        Note:
            This method is asynchronous (it queries the database), unlike the
            synchronous ``get_stats`` on
            :class:`~telegram_menu_builder.storage.memory.MemoryStorage`.

        Returns:
            A mapping with the keys ``total_keys``, ``keys_with_ttl``,
            ``expired_keys`` and ``active_keys``.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the read fails.
        """
        self._ensure_open()

        now = _utcnow()
        expires_at = self._table.c.expires_at
        is_expired = (expires_at.is_not(None)) & (expires_at <= now)
        # One aggregate query: total rows, rows with a TTL, and expired rows.
        # The expired count uses SUM(CASE ...) rather than COUNT(...) FILTER so it
        # stays portable across MySQL/MariaDB, which do not support the FILTER clause.
        stmt = select(
            func.count().label("total"),
            func.count(expires_at).label("with_ttl"),
            func.coalesce(func.sum(case((is_expired, 1), else_=0)), 0).label("expired"),
        )
        try:
            async with self._engine.connect() as conn:
                result = await conn.execute(stmt)
                row = result.one()
        except SQLAlchemyError as exc:
            raise StorageError(f"Failed to compute stats: {exc}") from exc

        total = int(row.total)
        with_ttl = int(row.with_ttl)
        expired_count = int(row.expired)
        return {
            "total_keys": total,
            "keys_with_ttl": with_ttl,
            "expired_keys": expired_count,
            "active_keys": total - expired_count,
        }

    async def close(self) -> None:
        """Close the storage, disposing the engine only if it is owned.

        A borrowed engine (supplied via ``engine=``) is left intact so the
        caller can keep using it.
        """
        if self._closed:
            return
        if self._owns_engine:
            await self._engine.dispose()
        await super().close()

create_schema async 

create_schema() -> None

Create the backing table and indexes if they do not exist.

This is idempotent: it uses checkfirst=True so it can be called repeatedly without error.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the DDL operation fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
async def create_schema(self) -> None:
    """Create the backing table and indexes if they do not exist.

    This is idempotent: it uses ``checkfirst=True`` so it can be called
    repeatedly without error.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the DDL operation fails.
    """
    self._ensure_open()
    try:
        async with self._engine.begin() as conn:
            await conn.run_sync(self._metadata.create_all, checkfirst=True)
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to create schema: {exc}") from exc

drop_schema async 

drop_schema() -> None

Drop the backing table if it exists.

This is idempotent thanks to checkfirst=True.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the DDL operation fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
async def drop_schema(self) -> None:
    """Drop the backing table if it exists.

    This is idempotent thanks to ``checkfirst=True``.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the DDL operation fails.
    """
    self._ensure_open()
    try:
        async with self._engine.begin() as conn:
            await conn.run_sync(self._metadata.drop_all, checkfirst=True)
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to drop schema: {exc}") from exc

set async 

set(key: str, data: dict[str, Any], ttl: int | None = None) -> None

Store (UPSERT) data under key with an optional TTL.

Re-setting the same key overwrites the existing value and refreshes (or clears) its expiry, matching the encoder's deterministic-key dedup.

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required
data dict[str, Any]

JSON-serializable dictionary to store.

 required
ttl int | None

Time-to-live in seconds (None = no expiration).

 None

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the write fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
    """Store (UPSERT) data under ``key`` with an optional TTL.

    Re-setting the same key overwrites the existing value and refreshes (or
    clears) its expiry, matching the encoder's deterministic-key dedup.

    Args:
        key: Unique identifier for the data.
        data: JSON-serializable dictionary to store.
        ttl: Time-to-live in seconds (``None`` = no expiration).

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the write fails.
    """
    self._ensure_open()

    expires_at = _utcnow() + datetime.timedelta(seconds=ttl) if ttl is not None else None
    values = {"key": key, "value": data, "expires_at": expires_at}

    try:
        stmt = self._build_upsert(values)
        if stmt is None:
            # Dialect without native UPSERT: fall back to DELETE-then-INSERT.
            await self._fallback_set(values)
        else:
            async with self._engine.begin() as conn:
                await conn.execute(stmt)
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to set key {key!r}: {exc}") from exc

add async 

add(key: str, data: dict[str, Any], ttl: int | None = None) -> bool

Atomically store data under key only if it is absent.

The whole operation runs inside a SINGLE transaction (async with self._engine.begin()) so it is atomic against concurrent callers and the engine's connection pool. Within that transaction it:

  1. DELETEs any row for key whose TTL has already elapsed (expires_at IS NOT NULL AND expires_at <= now), freeing an expired claim so it can be reclaimed; and then
  2. runs a dialect INSERT that no-ops on a primary-key conflict, mirroring the dialect branches of :meth:_build_upsert: PostgreSQL/SQLite use INSERT ... ON CONFLICT DO NOTHING; MySQL/MariaDB use INSERT IGNORE; any other dialect falls back to a plain INSERT and treats an :class:~sqlalchemy.exc.IntegrityError as a lost race.

The boolean result is derived from the INSERT's affected-row count: a row was inserted (won the claim) iff rowcount > 0.

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required
data dict[str, Any]

JSON-serializable dictionary to store.

 required
ttl int | None

Time-to-live in seconds (None = no expiration).

 None

Returns:

Type Description
bool

True if this call stored the value, False if a live
bool

(non-expired) row already existed for the key.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the write fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
    """Atomically store ``data`` under ``key`` only if it is absent.

    The whole operation runs inside a SINGLE transaction
    (``async with self._engine.begin()``) so it is atomic against concurrent
    callers and the engine's connection pool. Within that transaction it:

    1. DELETEs any row for ``key`` whose TTL has already elapsed
       (``expires_at IS NOT NULL AND expires_at <= now``), freeing an expired
       claim so it can be reclaimed; and then
    2. runs a dialect INSERT that no-ops on a primary-key conflict, mirroring
       the dialect branches of :meth:`_build_upsert`: PostgreSQL/SQLite use
       ``INSERT ... ON CONFLICT DO NOTHING``; MySQL/MariaDB use
       ``INSERT IGNORE``; any other dialect falls back to a plain INSERT and
       treats an :class:`~sqlalchemy.exc.IntegrityError` as a lost race.

    The boolean result is derived from the INSERT's affected-row count: a row
    was inserted (won the claim) iff ``rowcount > 0``.

    Args:
        key: Unique identifier for the data.
        data: JSON-serializable dictionary to store.
        ttl: Time-to-live in seconds (``None`` = no expiration).

    Returns:
        ``True`` if this call stored the value, ``False`` if a live
        (non-expired) row already existed for the key.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the write fails.
    """
    self._ensure_open()

    now = _utcnow()
    expires_at = now + datetime.timedelta(seconds=ttl) if ttl is not None else None
    values = {"key": key, "value": data, "expires_at": expires_at}

    try:
        async with self._engine.begin() as conn:
            # Free an expired claim first so it can be reclaimed in this txn.
            expires_col = self._table.c.expires_at
            await conn.execute(
                delete(self._table).where(
                    (self._table.c.key == key)
                    & (expires_col.is_not(None))
                    & (expires_col <= now)
                )
            )

            stmt = self._build_insert_ignore(values)
            if stmt is None:
                # Dialect without a native no-op insert: try a plain INSERT and
                # treat a primary-key collision as a lost race.
                try:
                    result = await conn.execute(self._table.insert().values(**values))
                except IntegrityError:
                    return False
                return result.rowcount > 0

            result = await conn.execute(stmt)
            return result.rowcount > 0
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to add key {key!r}: {exc}") from exc

get async 

get(key: str) -> dict[str, Any] | None

Retrieve the (non-expired) value stored under key.

Expiry is filtered lazily at query time; an expired row is treated as missing and None is returned (the row is not eagerly deleted here).

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required

Returns:

Type Description
dict[str, Any] | None

The stored dictionary, or None if missing or expired.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the read fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
async def get(self, key: str) -> dict[str, Any] | None:
    """Retrieve the (non-expired) value stored under ``key``.

    Expiry is filtered lazily at query time; an expired row is treated as
    missing and ``None`` is returned (the row is not eagerly deleted here).

    Args:
        key: Unique identifier for the data.

    Returns:
        The stored dictionary, or ``None`` if missing or expired.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the read fails.
    """
    self._ensure_open()

    now = _utcnow()
    stmt = select(self._table.c.value).where(
        (self._table.c.key == key) & self._not_expired(now)
    )
    try:
        async with self._engine.connect() as conn:
            result = await conn.execute(stmt)
            raw = result.scalar_one_or_none()
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to get key {key!r}: {exc}") from exc

    if raw is None:
        return None
    return cast("dict[str, Any]", raw)

delete async 

delete(key: str) -> bool

Delete the row stored under key.

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required

Returns:

Type Description
bool

True if a row was deleted, False if the key did not exist.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the delete fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
async def delete(self, key: str) -> bool:
    """Delete the row stored under ``key``.

    Args:
        key: Unique identifier for the data.

    Returns:
        ``True`` if a row was deleted, ``False`` if the key did not exist.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the delete fails.
    """
    self._ensure_open()

    stmt = delete(self._table).where(self._table.c.key == key)
    try:
        async with self._engine.begin() as conn:
            result = await conn.execute(stmt)
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to delete key {key!r}: {exc}") from exc

    return result.rowcount > 0

exists async 

exists(key: str) -> bool

Check whether a non-expired row exists under key.

Parameters:

Name Type Description Default
key str

Unique identifier to check.

 required

Returns:

Type Description
bool

True if the key exists and has not expired.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the read fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
async def exists(self, key: str) -> bool:
    """Check whether a non-expired row exists under ``key``.

    Args:
        key: Unique identifier to check.

    Returns:
        ``True`` if the key exists and has not expired.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the read fails.
    """
    self._ensure_open()

    now = _utcnow()
    stmt = select(self._table.c.key).where((self._table.c.key == key) & self._not_expired(now))
    try:
        async with self._engine.connect() as conn:
            result = await conn.execute(stmt)
            row = result.first()
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to check existence of key {key!r}: {exc}") from exc

    return row is not None

clear async 

clear() -> None

Delete every row in the backing table.

Warning

This removes ALL stored callbacks. Use with care in production.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the delete fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
async def clear(self) -> None:
    """Delete every row in the backing table.

    Warning:
        This removes ALL stored callbacks. Use with care in production.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the delete fails.
    """
    self._ensure_open()

    try:
        async with self._engine.begin() as conn:
            await conn.execute(delete(self._table))
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to clear storage: {exc}") from exc

keys async 

keys(pattern: str | None = None) -> list[str]

Return all non-expired keys, optionally filtered by a glob pattern.

The glob pattern is translated to a SQL LIKE clause and pushed down to the database.

Note

This diverges from :class:MemoryStorage's fnmatch semantics: SQL LIKE supports only */? (mapped to %/_) and has no [seq] character-class wildcards.

Parameters:

Name Type Description Default
pattern str | None

Optional glob pattern using * and ?.

 None

Returns:

Type Description
list[str]

A list of matching, non-expired keys.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the read fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
async def keys(self, pattern: str | None = None) -> list[str]:
    """Return all non-expired keys, optionally filtered by a glob pattern.

    The glob ``pattern`` is translated to a SQL ``LIKE`` clause and pushed
    down to the database.

    Note:
        This diverges from :class:`MemoryStorage`'s ``fnmatch`` semantics:
        SQL ``LIKE`` supports only ``*``/``?`` (mapped to ``%``/``_``) and
        has no ``[seq]`` character-class wildcards.

    Args:
        pattern: Optional glob pattern using ``*`` and ``?``.

    Returns:
        A list of matching, non-expired keys.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the read fails.
    """
    self._ensure_open()

    now = _utcnow()
    condition: ColumnElement[bool] = self._not_expired(now)
    if pattern is not None:
        like_clause: BinaryExpression[bool] = self._table.c.key.like(
            _glob_to_like(pattern), escape="\\"
        )
        condition = condition & like_clause

    stmt = select(self._table.c.key).where(condition)
    try:
        async with self._engine.connect() as conn:
            result = await conn.execute(stmt)
            rows = result.scalars().all()
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to list keys: {exc}") from exc

    return list(rows)

cleanup_expired async 

cleanup_expired() -> int

Delete all rows whose TTL has elapsed.

Returns:

Type Description
int

The number of expired rows removed.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the delete fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
async def cleanup_expired(self) -> int:
    """Delete all rows whose TTL has elapsed.

    Returns:
        The number of expired rows removed.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the delete fails.
    """
    self._ensure_open()

    now = _utcnow()
    expires_at = self._table.c.expires_at
    stmt = delete(self._table).where((expires_at.is_not(None)) & (expires_at <= now))
    try:
        async with self._engine.begin() as conn:
            result = await conn.execute(stmt)
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to clean up expired keys: {exc}") from exc

    return result.rowcount

get_stats async 

get_stats() -> dict[str, int]

Return counts describing the stored keys.

Note

This method is asynchronous (it queries the database), unlike the synchronous get_stats on :class:~telegram_menu_builder.storage.memory.MemoryStorage.

Returns:

Type Description
dict[str, int]

A mapping with the keys total_keys, keys_with_ttl,
dict[str, int]

expired_keys and active_keys.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the read fails.
Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
async def get_stats(self) -> dict[str, int]:
    """Return counts describing the stored keys.

    Note:
        This method is asynchronous (it queries the database), unlike the
        synchronous ``get_stats`` on
        :class:`~telegram_menu_builder.storage.memory.MemoryStorage`.

    Returns:
        A mapping with the keys ``total_keys``, ``keys_with_ttl``,
        ``expired_keys`` and ``active_keys``.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the read fails.
    """
    self._ensure_open()

    now = _utcnow()
    expires_at = self._table.c.expires_at
    is_expired = (expires_at.is_not(None)) & (expires_at <= now)
    # One aggregate query: total rows, rows with a TTL, and expired rows.
    # The expired count uses SUM(CASE ...) rather than COUNT(...) FILTER so it
    # stays portable across MySQL/MariaDB, which do not support the FILTER clause.
    stmt = select(
        func.count().label("total"),
        func.count(expires_at).label("with_ttl"),
        func.coalesce(func.sum(case((is_expired, 1), else_=0)), 0).label("expired"),
    )
    try:
        async with self._engine.connect() as conn:
            result = await conn.execute(stmt)
            row = result.one()
    except SQLAlchemyError as exc:
        raise StorageError(f"Failed to compute stats: {exc}") from exc

    total = int(row.total)
    with_ttl = int(row.with_ttl)
    expired_count = int(row.expired)
    return {
        "total_keys": total,
        "keys_with_ttl": with_ttl,
        "expired_keys": expired_count,
        "active_keys": total - expired_count,
    }

close async 

close() -> None

Close the storage, disposing the engine only if it is owned.

A borrowed engine (supplied via engine=) is left intact so the caller can keep using it.

Source code in src/telegram_menu_builder/storage/sqlalchemy.py 
674
675
676
677
678
679
680
681
682
683
684
async def close(self) -> None:
    """Close the storage, disposing the engine only if it is owned.

    A borrowed engine (supplied via ``engine=``) is left intact so the
    caller can keep using it.
    """
    if self._closed:
        return
    if self._owns_engine:
        await self._engine.dispose()
    await super().close()

RedisStorage

telegram_menu_builder.storage.redis.RedisStorage

Bases: BaseStorage

Async storage backend using redis-py, compatible with Redis and Valkey.

Callback payloads are stored as JSON strings under a configurable key namespace. The backend relies on the server to enforce TTL natively (via the SET ... EX option), so there is no client-side expiry bookkeeping.

Valkey support

Valkey is wire-compatible with Redis, so this same class connects to a Valkey server unchanged -- just pass a redis://valkey-host URL. A duck-typed valkey.asyncio client can also be supplied via client=.

Client ownership

If constructed from a url the backend creates and OWNS the client and closes it on :meth:close. If an existing client is supplied the backend BORROWS it and never closes it.

Namespacing

Every key is prefixed with namespace (default "tmb:"). All operations -- including :meth:clear and :meth:keys -- are scoped to that prefix, so multiple stores can safely share one server/database and :meth:clear never issues FLUSHDB.

Example

store = RedisStorage(url="redis://localhost:6379/0") await store.set("k", {"h": "menu", "p": {"x": 1}}) await store.get("k") {'h': 'menu', 'p': {'x': 1}} await store.close()

Note

Unlike :class:~telegram_menu_builder.storage.memory.MemoryStorage, this backend exposes no cleanup_expired or get_stats: Redis/Valkey enforce TTL natively, and server-side introspection is available through the INFO and DBSIZE commands on :attr:client.

Source code in src/telegram_menu_builder/storage/redis.py 
 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
302
303
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
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
class RedisStorage(BaseStorage):
    """Async storage backend using redis-py, compatible with Redis and Valkey.

    Callback payloads are stored as JSON strings under a configurable key
    namespace. The backend relies on the server to enforce TTL natively (via the
    ``SET ... EX`` option), so there is no client-side expiry bookkeeping.

    Valkey support:
        Valkey is wire-compatible with Redis, so this same class connects to a
        Valkey server unchanged -- just pass a ``redis://valkey-host`` URL. A
        duck-typed ``valkey.asyncio`` client can also be supplied via ``client=``.

    Client ownership:
        If constructed from a ``url`` the backend creates and OWNS the client and
        closes it on :meth:`close`. If an existing ``client`` is supplied the
        backend BORROWS it and never closes it.

    Namespacing:
        Every key is prefixed with ``namespace`` (default ``"tmb:"``). All
        operations -- including :meth:`clear` and :meth:`keys` -- are scoped to
        that prefix, so multiple stores can safely share one server/database and
        :meth:`clear` never issues ``FLUSHDB``.

    Example:
        >>> store = RedisStorage(url="redis://localhost:6379/0")
        >>> await store.set("k", {"h": "menu", "p": {"x": 1}})
        >>> await store.get("k")
        {'h': 'menu', 'p': {'x': 1}}
        >>> await store.close()

    Note:
        Unlike :class:`~telegram_menu_builder.storage.memory.MemoryStorage`, this
        backend exposes no ``cleanup_expired`` or ``get_stats``: Redis/Valkey
        enforce TTL natively, and server-side introspection is available through
        the ``INFO`` and ``DBSIZE`` commands on :attr:`client`.
    """

    def __init__(
        self,
        url: str | None = None,
        *,
        client: Redis | None = None,
        namespace: str = "tmb:",
    ) -> None:
        """Initialize the Redis/Valkey storage backend.

        Exactly one of ``url`` or ``client`` must be supplied.

        Args:
            url: A redis-py connection URL (e.g. ``"redis://localhost:6379/0"``
                or ``"rediss://..."`` for TLS). A Valkey server is reached via a
                ``redis://valkey-host`` URL. When given, a client is created and
                owned by this instance.
            client: An existing async client to borrow -- a
                :class:`redis.asyncio.Redis` or any duck-compatible async client
                (e.g. ``valkey.asyncio``). When given, it is NOT closed on
                :meth:`close`.
            namespace: Key prefix applied to every stored key (default
                ``"tmb:"``). Scopes :meth:`clear` and :meth:`keys`.

        Raises:
            ValueError: If neither or both of ``url`` and ``client`` are provided.
        """
        super().__init__()

        if (url is None) == (client is None):
            msg = "Exactly one of 'url' or 'client' must be provided (got both or neither)."
            raise ValueError(msg)

        self._client: Redis
        self._owns_client: bool
        if url is not None:
            # redis-py types ``Redis.from_url`` with ``**kwargs: Unknown``, which
            # makes pyright treat even the bare attribute access as partially
            # unknown. Reach it through an ``Any`` reference to the class, then cast
            # the result back to ``Redis``: mypy does not flag this cast as
            # redundant because its source is ``Any``.
            redis_cls: Any = Redis
            self._client = cast("Redis", redis_cls.from_url(url))
            self._owns_client = True
        else:
            # client is not None here (guaranteed by the XOR check above).
            assert client is not None
            self._client = client
            self._owns_client = False

        self._ns = namespace

    @property
    def client(self) -> Redis:
        """Return the underlying redis-py async client.

        Exposed for advanced use and inspection (e.g. ``await store.client.ttl``
        or ``await store.client.info()``).

        Returns:
            The wrapped :class:`redis.asyncio.Redis` client.
        """
        return self._client

    def _k(self, key: str) -> str:
        """Prefix a caller-supplied key with the configured namespace.

        Args:
            key: The unprefixed key as seen by callers.

        Returns:
            The fully namespaced key sent to the server.
        """
        return f"{self._ns}{key}"

    def _to_str(self, value: Any) -> str:
        """Decode a redis-py value to ``str``, handling bytes responses.

        redis-py may return ``bytes`` or ``str`` depending on the client's
        ``decode_responses`` setting; this normalises both to ``str``.

        Args:
            value: A key returned by ``scan_iter`` (``str`` or ``bytes``).

        Returns:
            The value as a ``str``.
        """
        return value.decode() if isinstance(value, (bytes, bytearray)) else value

    async def _scan(self, match: str) -> AsyncIterator[str]:
        """Yield namespaced keys matching ``match`` via ``SCAN``.

        Wraps ``scan_iter`` and normalises each yielded key to ``str``, containing
        redis-py's loosely typed (``Unknown``) async iterator behind a typed
        boundary.

        Args:
            match: A fully namespaced ``SCAN MATCH`` glob pattern.

        Yields:
            Each matching key as a ``str`` (still namespace-prefixed).
        """
        # redis-py types ``scan_iter`` as ``AsyncIterator[Unknown]`` and even its
        # attribute access as partially unknown. Reach it through an ``Any``
        # reference so the yielded values stay contained behind this helper rather
        # than leaking ``Unknown`` into the call sites.
        client: Any = self._client
        iterator: AsyncIterator[Any] = client.scan_iter(match=match)
        async for raw in iterator:
            yield self._to_str(raw)

    async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
        """Store ``data`` under ``key`` with an optional TTL.

        The value is serialized to compact JSON. Re-setting the same key
        overwrites the existing value (and resets or clears its expiry to match
        ``ttl``), exactly what the encoder's deterministic-key dedup needs.

        Args:
            key: Unique identifier for the data.
            data: JSON-serializable dictionary to store.
            ttl: Time-to-live in seconds (``None`` = no expiration; the key is
                persistent). The server enforces the expiry.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the write fails.
        """
        self._ensure_open()
        try:
            await self._client.set(self._k(key), json.dumps(data, separators=(",", ":")), ex=ttl)
        except RedisError as exc:
            raise StorageError(f"Failed to set key {key!r}: {exc}") from exc

    async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
        """Atomically store ``data`` under ``key`` only if it is absent.

        Implemented with a single ``SET ... NX [EX]`` command, which the server
        applies atomically: the value is written only when the key does not
        already exist. Because an expired key is gone server-side, ``NX``
        succeeds again once its TTL has elapsed (expired keys are reclaimable),
        and concurrent callers see exactly one ``True``.

        Args:
            key: Unique identifier for the data.
            data: JSON-serializable dictionary to store.
            ttl: Time-to-live in seconds (``None`` = no expiration). The server
                enforces the expiry.

        Returns:
            ``True`` if this call stored the value, ``False`` if a live
            (non-expired) value already existed for the key.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the write fails.
        """
        self._ensure_open()
        try:
            return bool(
                await self._client.set(
                    self._k(key),
                    json.dumps(data, separators=(",", ":")),
                    nx=True,
                    ex=ttl,
                )
            )
        except RedisError as exc:
            raise StorageError(f"Failed to add key {key!r}: {exc}") from exc

    async def get(self, key: str) -> dict[str, Any] | None:
        """Retrieve the value stored under ``key``.

        Expired keys are absent server-side, so this returns ``None`` for them
        without raising.

        Args:
            key: Unique identifier for the data.

        Returns:
            A fresh dictionary parsed from storage, or ``None`` if the key is
            missing or expired.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the read fails.
        """
        self._ensure_open()
        try:
            raw = await self._client.get(self._k(key))
        except RedisError as exc:
            raise StorageError(f"Failed to get key {key!r}: {exc}") from exc

        if raw is None:
            return None
        # json.loads accepts both str and bytes; it returns a fresh dict, so the
        # stored value is naturally isolated from callers.
        return cast("dict[str, Any]", json.loads(raw))

    async def delete(self, key: str) -> bool:
        """Delete the value stored under ``key``.

        Args:
            key: Unique identifier for the data.

        Returns:
            ``True`` if a key was removed, ``False`` if it did not exist.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the delete fails.
        """
        self._ensure_open()
        try:
            return bool(await self._client.delete(self._k(key)))
        except RedisError as exc:
            raise StorageError(f"Failed to delete key {key!r}: {exc}") from exc

    async def exists(self, key: str) -> bool:
        """Check whether ``key`` exists (and has not expired).

        Args:
            key: Unique identifier to check.

        Returns:
            ``True`` if the key exists server-side.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the read fails.
        """
        self._ensure_open()
        try:
            return bool(await self._client.exists(self._k(key)))
        except RedisError as exc:
            raise StorageError(f"Failed to check existence of key {key!r}: {exc}") from exc

    async def clear(self) -> None:
        """Delete every key in this store's namespace.

        Implemented with a namespace-scoped ``SCAN MATCH`` followed by ``DELETE``;
        it NEVER issues ``FLUSHDB``, so other namespaces sharing the same
        server/database are left untouched.

        Warning:
            This removes ALL callbacks stored under this namespace. Use with care
            in production.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the operation fails.
        """
        self._ensure_open()
        try:
            keys = [k async for k in self._scan(f"{self._ns}*")]
            if keys:
                await self._client.delete(*keys)
        except RedisError as exc:
            raise StorageError(f"Failed to clear storage: {exc}") from exc

    async def keys(self, pattern: str | None = None) -> list[str]:
        """Return the namespace's keys, optionally filtered by a glob pattern.

        The pattern is applied via ``SCAN MATCH`` and the namespace prefix is
        stripped from the results so callers see the keys they supplied.

        Note:
            ``SCAN MATCH`` uses Redis/Valkey glob syntax, which differs from
            Python's :mod:`fnmatch` (used by ``MemoryStorage``): character-class
            negation is written ``[^...]`` here rather than ``[!...]``.

        Args:
            pattern: Optional glob pattern (without the namespace prefix).

        Returns:
            A list of matching keys with the namespace prefix removed.

        Raises:
            RuntimeError: If the storage is closed.
            StorageError: If the read fails.
        """
        self._ensure_open()
        match = f"{self._ns}{pattern}" if pattern is not None else f"{self._ns}*"
        try:
            found = [k async for k in self._scan(match)]
        except RedisError as exc:
            raise StorageError(f"Failed to list keys: {exc}") from exc

        return [k.removeprefix(self._ns) for k in found]

    async def close(self) -> None:
        """Close the storage, closing the client only if it is owned.

        A borrowed client (supplied via ``client=``) is left open so the caller
        can keep using it.
        """
        if self._closed:
            return
        if self._owns_client:
            await self._client.aclose()
        await super().close()

client property 

client: Redis

Return the underlying redis-py async client.

Exposed for advanced use and inspection (e.g. await store.client.ttl or await store.client.info()).

Returns:

Type Description
Redis

The wrapped :class:redis.asyncio.Redis client.

set async 

set(key: str, data: dict[str, Any], ttl: int | None = None) -> None

Store data under key with an optional TTL.

The value is serialized to compact JSON. Re-setting the same key overwrites the existing value (and resets or clears its expiry to match ttl), exactly what the encoder's deterministic-key dedup needs.

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required
data dict[str, Any]

JSON-serializable dictionary to store.

 required
ttl int | None

Time-to-live in seconds (None = no expiration; the key is persistent). The server enforces the expiry.

 None

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the write fails.
Source code in src/telegram_menu_builder/storage/redis.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
async def set(self, key: str, data: dict[str, Any], ttl: int | None = None) -> None:
    """Store ``data`` under ``key`` with an optional TTL.

    The value is serialized to compact JSON. Re-setting the same key
    overwrites the existing value (and resets or clears its expiry to match
    ``ttl``), exactly what the encoder's deterministic-key dedup needs.

    Args:
        key: Unique identifier for the data.
        data: JSON-serializable dictionary to store.
        ttl: Time-to-live in seconds (``None`` = no expiration; the key is
            persistent). The server enforces the expiry.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the write fails.
    """
    self._ensure_open()
    try:
        await self._client.set(self._k(key), json.dumps(data, separators=(",", ":")), ex=ttl)
    except RedisError as exc:
        raise StorageError(f"Failed to set key {key!r}: {exc}") from exc

add async 

add(key: str, data: dict[str, Any], ttl: int | None = None) -> bool

Atomically store data under key only if it is absent.

Implemented with a single SET ... NX [EX] command, which the server applies atomically: the value is written only when the key does not already exist. Because an expired key is gone server-side, NX succeeds again once its TTL has elapsed (expired keys are reclaimable), and concurrent callers see exactly one True.

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required
data dict[str, Any]

JSON-serializable dictionary to store.

 required
ttl int | None

Time-to-live in seconds (None = no expiration). The server enforces the expiry.

 None

Returns:

Type Description
bool

True if this call stored the value, False if a live
bool

(non-expired) value already existed for the key.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the write fails.
Source code in src/telegram_menu_builder/storage/redis.py 
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
async def add(self, key: str, data: dict[str, Any], ttl: int | None = None) -> bool:
    """Atomically store ``data`` under ``key`` only if it is absent.

    Implemented with a single ``SET ... NX [EX]`` command, which the server
    applies atomically: the value is written only when the key does not
    already exist. Because an expired key is gone server-side, ``NX``
    succeeds again once its TTL has elapsed (expired keys are reclaimable),
    and concurrent callers see exactly one ``True``.

    Args:
        key: Unique identifier for the data.
        data: JSON-serializable dictionary to store.
        ttl: Time-to-live in seconds (``None`` = no expiration). The server
            enforces the expiry.

    Returns:
        ``True`` if this call stored the value, ``False`` if a live
        (non-expired) value already existed for the key.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the write fails.
    """
    self._ensure_open()
    try:
        return bool(
            await self._client.set(
                self._k(key),
                json.dumps(data, separators=(",", ":")),
                nx=True,
                ex=ttl,
            )
        )
    except RedisError as exc:
        raise StorageError(f"Failed to add key {key!r}: {exc}") from exc

get async 

get(key: str) -> dict[str, Any] | None

Retrieve the value stored under key.

Expired keys are absent server-side, so this returns None for them without raising.

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required

Returns:

Type Description
dict[str, Any] | None

A fresh dictionary parsed from storage, or None if the key is
dict[str, Any] | None

missing or expired.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the read fails.
Source code in src/telegram_menu_builder/storage/redis.py 
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
async def get(self, key: str) -> dict[str, Any] | None:
    """Retrieve the value stored under ``key``.

    Expired keys are absent server-side, so this returns ``None`` for them
    without raising.

    Args:
        key: Unique identifier for the data.

    Returns:
        A fresh dictionary parsed from storage, or ``None`` if the key is
        missing or expired.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the read fails.
    """
    self._ensure_open()
    try:
        raw = await self._client.get(self._k(key))
    except RedisError as exc:
        raise StorageError(f"Failed to get key {key!r}: {exc}") from exc

    if raw is None:
        return None
    # json.loads accepts both str and bytes; it returns a fresh dict, so the
    # stored value is naturally isolated from callers.
    return cast("dict[str, Any]", json.loads(raw))

delete async 

delete(key: str) -> bool

Delete the value stored under key.

Parameters:

Name Type Description Default
key str

Unique identifier for the data.

 required

Returns:

Type Description
bool

True if a key was removed, False if it did not exist.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the delete fails.
Source code in src/telegram_menu_builder/storage/redis.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
async def delete(self, key: str) -> bool:
    """Delete the value stored under ``key``.

    Args:
        key: Unique identifier for the data.

    Returns:
        ``True`` if a key was removed, ``False`` if it did not exist.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the delete fails.
    """
    self._ensure_open()
    try:
        return bool(await self._client.delete(self._k(key)))
    except RedisError as exc:
        raise StorageError(f"Failed to delete key {key!r}: {exc}") from exc

exists async 

exists(key: str) -> bool

Check whether key exists (and has not expired).

Parameters:

Name Type Description Default
key str

Unique identifier to check.

 required

Returns:

Type Description
bool

True if the key exists server-side.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the read fails.
Source code in src/telegram_menu_builder/storage/redis.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
async def exists(self, key: str) -> bool:
    """Check whether ``key`` exists (and has not expired).

    Args:
        key: Unique identifier to check.

    Returns:
        ``True`` if the key exists server-side.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the read fails.
    """
    self._ensure_open()
    try:
        return bool(await self._client.exists(self._k(key)))
    except RedisError as exc:
        raise StorageError(f"Failed to check existence of key {key!r}: {exc}") from exc

clear async 

clear() -> None

Delete every key in this store's namespace.

Implemented with a namespace-scoped SCAN MATCH followed by DELETE; it NEVER issues FLUSHDB, so other namespaces sharing the same server/database are left untouched.

Warning

This removes ALL callbacks stored under this namespace. Use with care in production.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the operation fails.
Source code in src/telegram_menu_builder/storage/redis.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
async def clear(self) -> None:
    """Delete every key in this store's namespace.

    Implemented with a namespace-scoped ``SCAN MATCH`` followed by ``DELETE``;
    it NEVER issues ``FLUSHDB``, so other namespaces sharing the same
    server/database are left untouched.

    Warning:
        This removes ALL callbacks stored under this namespace. Use with care
        in production.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the operation fails.
    """
    self._ensure_open()
    try:
        keys = [k async for k in self._scan(f"{self._ns}*")]
        if keys:
            await self._client.delete(*keys)
    except RedisError as exc:
        raise StorageError(f"Failed to clear storage: {exc}") from exc

keys async 

keys(pattern: str | None = None) -> list[str]

Return the namespace's keys, optionally filtered by a glob pattern.

The pattern is applied via SCAN MATCH and the namespace prefix is stripped from the results so callers see the keys they supplied.

Note

SCAN MATCH uses Redis/Valkey glob syntax, which differs from Python's :mod:fnmatch (used by MemoryStorage): character-class negation is written [^...] here rather than [!...].

Parameters:

Name Type Description Default
pattern str | None

Optional glob pattern (without the namespace prefix).

 None

Returns:

Type Description
list[str]

A list of matching keys with the namespace prefix removed.

Raises:

Type Description
RuntimeError

If the storage is closed.
StorageError

If the read fails.
Source code in src/telegram_menu_builder/storage/redis.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
async def keys(self, pattern: str | None = None) -> list[str]:
    """Return the namespace's keys, optionally filtered by a glob pattern.

    The pattern is applied via ``SCAN MATCH`` and the namespace prefix is
    stripped from the results so callers see the keys they supplied.

    Note:
        ``SCAN MATCH`` uses Redis/Valkey glob syntax, which differs from
        Python's :mod:`fnmatch` (used by ``MemoryStorage``): character-class
        negation is written ``[^...]`` here rather than ``[!...]``.

    Args:
        pattern: Optional glob pattern (without the namespace prefix).

    Returns:
        A list of matching keys with the namespace prefix removed.

    Raises:
        RuntimeError: If the storage is closed.
        StorageError: If the read fails.
    """
    self._ensure_open()
    match = f"{self._ns}{pattern}" if pattern is not None else f"{self._ns}*"
    try:
        found = [k async for k in self._scan(match)]
    except RedisError as exc:
        raise StorageError(f"Failed to list keys: {exc}") from exc

    return [k.removeprefix(self._ns) for k in found]

close async 

close() -> None

Close the storage, closing the client only if it is owned.

A borrowed client (supplied via client=) is left open so the caller can keep using it.

Source code in src/telegram_menu_builder/storage/redis.py 
359
360
361
362
363
364
365
366
367
368
369
async def close(self) -> None:
    """Close the storage, closing the client only if it is owned.

    A borrowed client (supplied via ``client=``) is left open so the caller
    can keep using it.
    """
    if self._closed:
        return
    if self._owns_client:
        await self._client.aclose()
    await super().close()