peyk.fsm.context

FSM context and event-to-storage-key compatibility helpers.

Functions

build_storage_key(event, *, platform, bot_id)	Build a platform/bot-aware key from a normalized event.
conversation_key(event, *[, platform, bot_id])	Return a compatibility StorageKey for an event.

Classes

FSMContext()	Read and mutate state/data for one StorageKey.
FSMStrategy(*values)	Select which event identities share one FSM conversation.

class peyk.fsm.context.FSMStrategy(*values)

Bases: str, Enum

Select which event identities share one FSM conversation.

USER_IN_CHAT = 'USER_IN_CHAT'

CHAT = 'CHAT'

GLOBAL_USER = 'GLOBAL_USER'

peyk.fsm.context.build_storage_key(event: object, *, platform: str, bot_id: int | str, strategy: str | FSMStrategy = FSMStrategy.USER_IN_CHAT, destiny: str = 'default') → StorageKey

Build a platform/bot-aware key from a normalized event.

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

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.context.FSMContext(storage: BaseStorage, key: StorageKey)

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

classmethod for_event(event: object, storage: BaseStorage, *, platform: str, bot_id: int | str, strategy: str | FSMStrategy = FSMStrategy.USER_IN_CHAT, destiny: str = 'default') → FSMContext

Create a context for a normalized event using explicit identity data.

async get_state() → str | None

Return the current state.

async set_state(state: State | str | None) → None

Set or clear the current state.

async get_data() → dict[str, object]

Return a copy of conversation data.

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

Replace conversation data.

async get_value(key: str, default: object = None) → object

Return one data value, or default when absent.

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

Merge mapping and keyword values into conversation data.

async clear() → None

Clear both state and conversation data.