peyk.fsm¶

Finite-state conversation support for Peyk.

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

peyk.fsm.build_storage_key(event: object, *, platform: str, bot_id: int | str, strategy: str | FSMStrategy = FSMStrategy.USER_IN_CHAT, destiny: str = 'default') → StorageKey[source]¶: Build a platform/bot-aware key from a normalized event.

peyk.fsm.conversation_key(event: object, *, platform: str | None = None, bot_id: int | str | None = None) → StorageKey[source]¶

Return a compatibility StorageKey for an event.

The old form inferred the platform from event.raw and had no bot ID. That form is deprecated; dispatcher integrations must pass both values.

class peyk.fsm.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.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.FSMContext(storage: BaseStorage, key: StorageKey)[source]¶

class peyk.fsm.FSMContext(event: object, storage: BaseStorage | LegacyStringStorage, *, platform: str | None = None, bot_id: int | str | None = None)

Bases: object

Read and mutate state/data for one StorageKey.

async clear() → None[source]¶: Clear both state and conversation data.

classmethod for_event(event: object, storage: BaseStorage, *, platform: str, bot_id: int | str, strategy: str | FSMStrategy = FSMStrategy.USER_IN_CHAT, destiny: str = 'default') → FSMContext[source]¶: Create a context for a normalized event using explicit identity data.

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

async get_state() → str | None[source]¶: Return the current state.

async get_value(key: str, default: object = None) → object[source]¶: Return one data value, or default when absent.

async set_data(data: Mapping[str, object]) → None[source]¶: Replace conversation data.

async set_state(state: State | str | None) → None[source]¶: Set or clear the current state.

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

class peyk.fsm.FSMStrategy(*values)[source]¶

Bases: str, Enum

Select which event identities share one FSM conversation.

USER_IN_CHAT = 'USER_IN_CHAT'¶

CHAT = 'CHAT'¶

GLOBAL_USER = 'GLOBAL_USER'¶

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

class peyk.fsm.State(state: str | None = None, group_name: str | None = None)[source]¶

Bases: object

Describe one named FSM state, optionally with an explicit group name.

property group: type[StatesGroup]¶: Return the owning state-group class.

set_parent(group: type[StatesGroup]) → None[source]¶: Bind this state to a StatesGroup class.

property state: str | None¶: Return the fully-qualified state name.

class peyk.fsm.StatesGroup[source]¶

Bases: object

Base class for declarative FSM state groups.

classmethod get_root() → type[StatesGroup][source]¶: Return the root group for a nested group.

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

`context`	FSM context and event-to-storage-key compatibility helpers.
`middleware`	Dispatcher middleware that injects FSM context and event isolation.
`state`	Declarative FSM states compatible with aiogram 3 semantics.
`storage`	FSM storage backends and event isolation.