26 lines
1002 B
Python
26 lines
1002 B
Python
import redis.asyncio import redis
|
|
from aiogram.fsm.storage.base import BaseStorage
|
|
from aiogram.fsm.storage.memory import MemoryStorage
|
|
from aiogram.fsm.storage.redis import RedisStorage
|
|
|
|
from .enums import StorageTypeEnum
|
|
from .settings import StorageSettings
|
|
|
|
|
|
def get_telegram_bot_fsm_storage(settings: StorageSettings) -> BaseStorage:
|
|
match settings.type:
|
|
case StorageTypeEnum.MEMORY.value:
|
|
return MemoryStorage()
|
|
case StorageTypeEnum.REDIS.value:
|
|
redis_connection_pool = redis.ConnectionPool(
|
|
host=settings.host,
|
|
port=settings.port,
|
|
db=settings.database,
|
|
password=settings.password,
|
|
max_connections=settings.max_connections,
|
|
)
|
|
redis_client = redis.Redis(connection_pool=redis_connection_pool)
|
|
return RedisStorage(redis=redis_client)
|
|
case _:
|
|
raise ValueError(f"Unsupported telegram bot storage type: {settings.type}")
|