SQLite storage

from aiogram_magick.sqlite import SqliteStorage

# By default, SqliteStorage is configured to:
#    - Commit changes on 30 minute idle and on shutdown;
#    - Cache states (up to 20 entries) and data (up to 10 entries);
#    - Ignore any exceptions;
#    - To avoid file corruptions on shutdown any `sqlite3.OperationalError`s
#      are printed using `traceback.print_exception` instead of raised normally.
dp = Dispatcher(storage=SqliteStorage('aiogram.sqlite'))

Configuration

Database path

SqliteStorage('/path/to/database/file.sqlite')

Commit frequency

By default, SqliteStorage commits changes (internally saved in __commit) on:

• the first request of changing a state or data;
• After idle_to_commit seconds of idling between either set_state(), set_data() or update_data();
• close().

Key factory

SqliteStorage does not provide an option to configure the key factory that is used for storing keys in the database. It also does not use aiogram.fsm.storage.base.DefaultKeyFactory nor bases on it. The reason is simple: DefaultKeyFactory does not provide a method to convert str back to StorageKey that can be used to save data between reloads.

__key_to_sqlite and __sqlite_to_key uses this format to work with StorageKeys:

<bot id>:<chat id>:<user id>

Warning

thread_id, business_connection_id and destiny fields of StorageKey are ignored.

API Reference

SQLite-based storage with caching & automatic commits for Aiogram 3.x.

SqliteStorage

SqliteStorage(filepath, idle_to_commit=1800)

Bases: BaseStorage

SQLite FSM storage.

Initalize a SQLite-based storage.

Parameters:

Name	Type	Description	Default
filepath	str	Path to the database file	required
idle_to_commit	int	Frequency of idling to commit changes (in seconds). It it notable to mention that regardless of this value all changes will be saved anyway when self.close() is called. Defaults to 1800.	1800

Source code in aiogram_magick/sqlite.py

def __init__(self, filepath: str, idle_to_commit: int = 1800):
    """Initalize a SQLite-based storage.

    Args:
        filepath (str): Path to the database file
        idle_to_commit (int): Frequency of idling to commit changes (in seconds). It it notable
            to mention that regardless of this value all changes will be saved anyway when
            `self.close()` is called. Defaults to `1800`.
    """
    self.filepath = filepath
    self.idle_to_commit = idle_to_commit

    self.__last_commit_ts = 0
    self.__commit: Dict[StorageKey, Dict[str, Union[StateType, Dict[str]]]] = {}

    self.__cache_state = SqliteStorageCache(capacity=20)
    self.__cache_data = SqliteStorageCache()

    self.__conn: Union[aiosqlite.Connection, None] = None

__cache_data instance-attribute

__cache_data = SqliteStorageCache()

__cache_state instance-attribute

__cache_state = SqliteStorageCache(capacity=20)

__commit instance-attribute

__commit = {}

__conn instance-attribute

__conn = None

__last_commit_ts instance-attribute

__last_commit_ts = 0

filepath instance-attribute

filepath = filepath

idle_to_commit instance-attribute

idle_to_commit = idle_to_commit

__connect async

__connect()

Source code in aiogram_magick/sqlite.py

async def __connect(self):
    if self.__conn:
        return

    aiosqlite.register_adapter(StorageKey, self.__key_to_sqlite)
    aiosqlite.register_converter('StorageKey', self.__sqlite_to_key)

    # NOTE: Can an aiosqlite connection be handled in a thread-safe way?
    self.__conn = await aiosqlite.connect(
        self.filepath, check_same_thread=False, detect_types=PARSE_DECLTYPES
    )
    self.__conn.isolation_level = None  # TODO: (Python 3.12 needed) Use .autocommit instead
    await self.__conn.executescript(
        'CREATE TABLE IF NOT EXISTS "aiogram_states"('
        '"Key"     StorageKey NOT NULL UNIQUE,'
        '"State"   VARCHAR(100) DEFAULT NULL,'
        '"Data"    TEXT,'
        'PRIMARY KEY("Key")'
        ');'
        'CREATE UNIQUE INDEX IF NOT EXISTS "aiogram_states_keys" ON '
        '"aiogram_states" ("Key");'
    )
    await self.__conn.commit()

__key_to_sqlite staticmethod

__key_to_sqlite(key)

Source code in aiogram_magick/sqlite.py

@staticmethod
def __key_to_sqlite(key: StorageKey) -> str:
    return f'{key.bot_id}:{key.chat_id}:{key.user_id}'

__sqlite_to_key staticmethod

__sqlite_to_key(value)

Source code in aiogram_magick/sqlite.py

@staticmethod
def __sqlite_to_key(value: str) -> StorageKey:
    parts = value.split(':')
    return StorageKey(bot_id=int(parts[0]), chat_id=int(parts[1]), user_id=int(parts[2]))

SqliteStorageCache dataclass

SqliteStorageCache(cache=list(), keys=list(), length=0, *, capacity=10)

SQLite cache dataclass.

cache class-attribute instance-attribute

cache = field(default_factory=list)

capacity class-attribute instance-attribute

capacity = field(default=10, compare=False, kw_only=True)

keys class-attribute instance-attribute

keys = field(default_factory=list)

length class-attribute instance-attribute

length = 0

add

add(key, value)

Add a new cache entry of key.

Parameters:

Name	Type	Description	Default
key	StorageKey	Cache key.	required
value	Union[StateType, Dict[str]]	Value	required

Source code in aiogram_magick/sqlite.py

def add(self, key: StorageKey, value: 'Union[StateType, Dict[str]]'):
    """Add a new cache entry of `key`.

    Args:
        key (StorageKey): Cache key.
        value (Union[StateType, Dict[str]]): Value
    """
    if self.length == self.capacity:
        self.keys.pop()
        self.cache.pop()
    else:
        self.length += 1

    self.keys.insert(0, key)
    self.cache.insert(0, value)

get

get(key)

Get the cache entry of key if it exists.

Parameters:

Name	Type	Description	Default
key	StorageKey	Cache key.	required

Returns:

Type	Description
Union[StateType, Dict[str], None]	Union[StateType, Dict[str], None]: State, data or None

Source code in aiogram_magick/sqlite.py

def get(self, key: StorageKey) -> 'Union[StateType, Dict[str], None]':
    """Get the cache entry of `key` if it exists.

    Args:
        key (StorageKey): Cache key.

    Returns:
        Union[StateType, Dict[str], None]: State, data or `None`
    """
    key_i = self.keys.index(key)
    self.keys.insert(0, self.keys.pop(key_i))
    self.cache.insert(0, self.cache.pop(key_i))
    return self.cache[0]

update

update(key, value)

Update the value for the cache entry of key.

Parameters:

Name	Type	Description	Default
key	StorageKey	Cache key.	required
value	Union[StateType, Dict[str]]	New value	required

Source code in aiogram_magick/sqlite.py

def update(self, key: StorageKey, value: 'Union[StateType, Dict[str]]'):
    """Update the value for the cache entry of `key`.

    Args:
        key (StorageKey): Cache key.
        value (Union[StateType, Dict[str]]): New value
    """
    key_i = self.keys.index(key)
    self.keys.insert(0, self.keys.pop(key_i))
    self.cache.insert(0, self.cache.pop(key_i))
    self.cache[0] = value