peyk.fsm.storage¶

FSM storage backends and event isolation.

class peyk.fsm.storage.BaseEventIsolation[source]¶

Bases: object

Serialize handlers that share one StorageKey.

async close() → None[source]¶: Close isolation resources.

lock(key: StorageKey) → AsyncIterator[None][source]¶: Return an async context manager for the event lock.

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

Bases: ABC

Abstract asynchronous FSM storage keyed by StorageKey.

abstractmethod async close() → None[source]¶: Performs the close operation for the FSM client.

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 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_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 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 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.

class peyk.fsm.storage.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.

class peyk.fsm.storage.DisabledEventIsolation[source]¶

Bases: BaseEventIsolation

Do not serialize events.

lock(key: StorageKey) → AsyncIterator[None][source]¶

Performs the lock operation for the FSM client.

Parameters:: key – Value used by this operation.
Returns:: Result produced by the FSM operation.

class peyk.fsm.storage.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.LegacyStorageAdapter(storage: LegacyStringStorage, *, key_builder: KeyBuilder | None = None)[source]¶

Bases: BaseStorage

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

async close() → None[source]¶: Performs the close operation for the FSM client.

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 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_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 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 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.

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

Bases: Protocol

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

async close() → None[source]¶: Performs the close operation for the FSM client.

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 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_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 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 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.

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

Bases: BaseStorage

Store FSM state and data in process memory.

async close() → None[source]¶: Release in-memory data and mark the backend closed.

async get_data(key: StorageKey | str) → dict[str, object][source]¶: Return a defensive copy of conversation data.

async get_state(key: StorageKey | str) → str | None[source]¶: Return the stored state for key.

async set_data(key: StorageKey | str, data: Mapping[str, object]) → None[source]¶: Replace conversation data with a defensive copy.

async set_state(key: StorageKey | str, state: str | None) → None[source]¶: Set or clear the stored state.

async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object][source]¶: Merge mapping and keyword values into conversation data.

class peyk.fsm.storage.RedisEventIsolation(redis: RedisLockClient, *, key_builder: KeyBuilder | None = None, lock_timeout: float | None = 30.0, blocking_timeout: float | None = None)[source]¶

Bases: BaseEventIsolation

Serialize events using Redis distributed locks.

async close() → None[source]¶: Performs the close operation for the FSM client.

lock(key: StorageKey) → AsyncIterator[None][source]¶

Performs the lock operation for the FSM client.

Parameters:: key – Value used by this operation.
Returns:: Result produced by the FSM operation.

Bases: BaseStorage

Persist FSM state/data in Redis with independent optional TTLs.

async close() → None[source]¶: Close the underlying Redis client when supported.

async get_data(key: StorageKey | str) → dict[str, object][source]¶: Decode JSON conversation data from Redis.

async get_state(key: StorageKey | str) → str | None[source]¶: Return a stored state string, if present.

async set_data(key: StorageKey | str, data: Mapping[str, object]) → None[source]¶: Replace JSON conversation data and apply data_ttl.

async set_state(key: StorageKey | str, state: str | None) → None[source]¶: Set or clear state, applying state_ttl when configured.

async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object][source]¶: Read, merge and persist conversation data.

class peyk.fsm.storage.SimpleEventIsolation[source]¶

Bases: BaseEventIsolation

Serialize by key and evict idle locks after release.

async close() → None[source]¶: Performs the close operation for the FSM client.

lock(key: StorageKey) → AsyncIterator[None][source]¶

Performs the lock operation for the FSM client.

Parameters:: key – Value used by this operation.
Returns:: Result produced by the FSM operation.

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¶

Modules

`base`	Storage contracts and key builders for the dispatcher FSM.
`isolation`	FSM event isolation primitives.
`memory`	In-process FSM storage.
`redis`	Optional Redis-backed FSM storage.