raito.utils.storages.sql.sqlalchemy module

class raito.utils.storages.sql.sqlalchemy.SQLAlchemyStorage(engine, session_factory=None, key_separator=':')[source]

Bases: BaseStorage

SQLAlchemy storage for FSM.

Parameters:

  • engine (AsyncEngine)

  • session_factory (async_sessionmaker[AsyncSession] | None)

  • key_separator (str)

async classmethod from_url(url, *, echo=False, pool_size=10, max_overflow=0, **kwargs)[source]

Create storage from database URL.

Parameters:

  • url (str) – Database URL

  • echo (bool) – Enable SQL logging

  • pool_size (int) – Connection pool size

  • max_overflow (int) – Max overflow connections

  • kwargs (Any)

Returns:

Configured storage instance

Return type:

BaseSQLAlchemyStorage

async get_state(key)[source]

Get key state.

Parameters:

key (StorageKey) – Storage key

Returns:

Current state

Return type:

str | None

async get_data(key)[source]

Get current data for key.

Parameters:

key (StorageKey) – Storage key

Returns:

Current data

Return type:

Dict[str, Any]

async update_data(key, data)[source]

Update data in the storage for key.

Parameters:

  • key (StorageKey) – Storage key

  • data (Dict[str, Any]) – Data to update

Returns:

Updated data

Return type:

Dict[str, Any]

async close()[source]

Close the storage.

Return type:

None

async migrate()[source]

Create tables if not exist.

Return type:

None