40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
import os
|
|
import redis.asyncio as redis
|
|
from dotenv import find_dotenv, load_dotenv
|
|
|
|
from Craft.database.sqlite import Sqlite
|
|
load_dotenv(find_dotenv())
|
|
|
|
|
|
class Redis:
|
|
def __init__(self):
|
|
self.redis_host: str = os.getenv("REDIS_HOST")
|
|
self.redis_port: int = os.getenv("REDIS_PORT")
|
|
self.redis_db: int = os.getenv("REDIS_DB")
|
|
self.redis_password: str = os.getenv("REDIS_PASSWORD")
|
|
|
|
self.redis_pool = redis.ConnectionPool(
|
|
host=self.redis_host,
|
|
port=self.redis_port,
|
|
db=self.redis_db,
|
|
password=self.redis_password
|
|
)
|
|
self.redis_instance = redis.Redis(connection_pool=self.redis_pool)
|
|
|
|
|
|
async def __aenter__(self):
|
|
return self.redis_instance
|
|
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
await self.redis_instance.close()
|
|
|
|
|
|
async def sync_data(self) -> None:
|
|
await self.redis_instance.flushdb()
|
|
async with Sqlite() as sqlite:
|
|
await sqlite.execute("SELECT * FROM Craft")
|
|
data = await sqlite.fetchall()
|
|
for row in data:
|
|
key1, key2, value_emoji, value_word, _ = row
|
|
self.redis_instance.hmset(f"CraftCached:{key1}:{key2}", {"emoji": value_emoji, "word": value_word}) |