peyk.fsm.storage.base

Storage contracts and key builders for the dispatcher FSM.

Functions

coerce_storage_key(key)

Convert a legacy string key while preserving old storage tests.

Classes

BaseStorage(*[, key_builder])

Abstract asynchronous FSM storage keyed by StorageKey.

DefaultKeyBuilder([prefix, separator, ...])

Build aiogram-shaped FSM keys with platform and bot namespaces.

KeyBuilder(*args, **kwargs)

Build backend keys from a StorageKey.

LegacyStorageAdapter(storage, *[, key_builder])

Adapt the pre-Phase-8 string-key storage API to StorageKey.

LegacyStringStorage(*args, **kwargs)

Protocol for the pre-Phase-8 string-key storage API.

StorageKey(platform, bot_id, chat_id, user_id)

Identify one FSM namespace without embedding a bot token.
class peyk.fsm.storage.base.StorageKey(platform: str, bot_id: int | str, chat_id: int | str | None, user_id: int | str | None, destiny: str = 'default')[source]

Bases: object

Identify one FSM namespace without embedding a bot token.

platform is deliberately part of the identity so equal numeric IDs on Telegram and Bale cannot share state. bot_id isolates two bots on the same platform.

platform: str
bot_id: int | str
chat_id: int | str | None
user_id: int | str | None
destiny: str
class peyk.fsm.storage.base.KeyBuilder(*args, **kwargs)[source]

Bases: Protocol

Build backend keys from a StorageKey.

build(key: StorageKey, part: str | None = None) → str[source]

Performs the build operation for the FSM client.

Parameters:

  • key – Value used by this operation.

  • part – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.storage.base.DefaultKeyBuilder(prefix: str = 'peyk', separator: str = ':', with_bot_id: bool = True, with_destiny: bool = True)[source]

Bases: object

Build aiogram-shaped FSM keys with platform and bot namespaces.

The default is equivalent to aiogram’s with_bot_id=True plus the additional platform segment required by Peyk’s multi-platform contract.

prefix: str
separator: str
with_bot_id: bool
with_destiny: bool
build(key: StorageKey, part: str | None = None) → str[source]

Performs the build operation for the FSM client.

Parameters:

  • key – Value used by this operation.

  • part – Value used by this operation.

Returns:

Result produced by the FSM operation.

peyk.fsm.storage.base.coerce_storage_key(key: StorageKey | str) → StorageKey[source]

Convert a legacy string key while preserving old storage tests.

New dispatcher code always passes StorageKey; string keys are a deprecated compatibility escape hatch and cannot provide bot isolation.

class peyk.fsm.storage.base.BaseStorage(*, key_builder: KeyBuilder | None = None)[source]

Bases: ABC

Abstract asynchronous FSM storage keyed by StorageKey.

abstractmethod async get_state(key: StorageKey | str) → str | None[source]

Retrieves state from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

abstractmethod async set_state(key: StorageKey | str, state: str | None) → None[source]

Updates state through the FSM API.

Parameters:

  • key – Value used by this operation.

  • state – Value used by this operation.

abstractmethod async get_data(key: StorageKey | str) → dict[str, object][source]

Retrieves data from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

abstractmethod async set_data(key: StorageKey | str, data: Mapping[str, object]) → None[source]

Updates data through the FSM API.

Parameters:

  • key – Value used by this operation.

  • data – Value used by this operation.

abstractmethod async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object][source]

Performs the update data operation for the FSM client.

Parameters:

  • key – Value used by this operation.

  • data – Value used by this operation.

Returns:

Result produced by the FSM operation.

abstractmethod async close() → None[source]

Performs the close operation for the FSM client.

class peyk.fsm.storage.base.LegacyStringStorage(*args, **kwargs)[source]

Bases: Protocol

Protocol for the pre-Phase-8 string-key storage API.

async get_state(key: str) → str | None[source]

Retrieves state from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async set_state(key: str, state: str | None) → None[source]

Updates state through the FSM API.

Parameters:

  • key – Value used by this operation.

  • state – Value used by this operation.

async get_data(key: str) → dict[str, object][source]

Retrieves data from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async set_data(key: str, data: Mapping[str, object]) → None[source]

Updates data through the FSM API.

Parameters:

  • key – Value used by this operation.

  • data – Value used by this operation.

async update_data(key: str, **kwargs: object) → dict[str, object][source]

Performs the update data operation for the FSM client.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async close() → None[source]

Performs the close operation for the FSM client.

class peyk.fsm.storage.base.LegacyStorageAdapter(storage: LegacyStringStorage, *, key_builder: KeyBuilder | None = None)[source]

Bases: BaseStorage

Adapt the pre-Phase-8 string-key storage API to StorageKey.

async get_state(key: StorageKey | str) → str | None[source]

Retrieves state from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async set_state(key: StorageKey | str, state: str | None) → None[source]

Updates state through the FSM API.

Parameters:

  • key – Value used by this operation.

  • state – Value used by this operation.

async get_data(key: StorageKey | str) → dict[str, object][source]

Retrieves data from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async set_data(key: StorageKey | str, data: Mapping[str, object]) → None[source]

Updates data through the FSM API.

Parameters:

  • key – Value used by this operation.

  • data – Value used by this operation.

async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object][source]

Performs the update data operation for the FSM client.

Parameters:

  • key – Value used by this operation.

  • data – Value used by this operation.

Returns:

Result produced by the FSM operation.

async close() → None[source]

Performs the close operation for the FSM client.