diff --git a/aiogram/fsm/storage/sqlite.py b/aiogram/fsm/storage/sqlite.py new file mode 100644 index 00000000..c5954973 --- /dev/null +++ b/aiogram/fsm/storage/sqlite.py @@ -0,0 +1,71 @@ +from aiosqlite import connect, Connection +from typing import Any, Dict, Mapping, Optional, cast + +from aiogram.fsm.storage.base import ( + BaseEventIsolation, + BaseStorage, + DefaultKeyBuilder, + KeyBuilder, + StateType, + StorageKey, +) + +SQLITE_FILENAME = "aiogram_fsm.sqlite" + + +class SqliteStorage(BaseStorage): + """ + SQLite storage required :code:`aiosqlite` package installed (:code:`pip install aiosqlite`) + """ + + def __init__( + self, + connection: Connection, + key_builder: Optional[KeyBuilder] = None, + ) -> None: + """ + Create an instance of :class:`SqliteStorage` with provided SQLite connection + + :param key_builder: builder that helps to convert contextual key to string + :param db_filename: name of the SQLite database file for FSM + """ + if key_builder is None: + key_builder = DefaultKeyBuilder() + self._key_builder = key_builder + self.connection = connection + + @classmethod + async def connect(cls, db_filename: str = SQLITE_FILENAME) -> Connection: + """ + Create an instance of :class:`SqliteStorage` with specifying the DB filename + + :param db_filename: for example :code:`aiogram_fsm.sqlite` + :param connection_kwargs: see :code:`motor` docs + :param kwargs: arguments to be passed to :class:`MongoStorage` + :return: an instance of :class:`MongoStorage` + """ + connection = await connect(db_filename) + await connection.execute(f'''CREATE TABLE IF NOT EXISTS aiogram_fsm ( + id TEXT PRIMARY KEY, + state BLOB, + data BLOB)''') + await connection.commit() + return cls(connection=connection) + + async def close(self) -> None: + await self.connection.close() + + async def set_state(self, key: StorageKey, state: StateType = None) -> None: + pass + + async def get_state(self, key: StorageKey) -> Optional[str]: + pass + + async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None: + pass + + async def get_data(self, key: StorageKey) -> Dict[str, Any]: + pass + + +