55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
"""
|
||
KeyvalDB - 键值数据库（简化版，仅使用 aiosqlite）
|
||
"""
|
||
|
||
import aiosqlite
|
||
from loguru import logger
|
||
from utils.singleton import Singleton
|
||
|
||
|
||
class KeyvalDB(metaclass=Singleton):
|
||
"""键值存储数据库"""
|
||
|
||
def __init__(self):
|
||
self.db_path = "database/keyval.db"
|
||
self.conn = None
|
||
|
||
async def initialize(self):
|
||
"""初始化数据库"""
|
||
self.conn = await aiosqlite.connect(self.db_path)
|
||
await self.conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS keyval (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT
|
||
)
|
||
""")
|
||
await self.conn.commit()
|
||
logger.debug("KeyvalDB 初始化完成")
|
||
|
||
async def set(self, key: str, value: str):
|
||
"""设置键值"""
|
||
await self.conn.execute(
|
||
"INSERT OR REPLACE INTO keyval (key, value) VALUES (?, ?)",
|
||
(key, value)
|
||
)
|
||
await self.conn.commit()
|
||
|
||
async def get(self, key: str):
|
||
"""获取值"""
|
||
cursor = await self.conn.execute(
|
||
"SELECT value FROM keyval WHERE key = ?",
|
||
(key,)
|
||
)
|
||
row = await cursor.fetchone()
|
||
return row[0] if row else None
|
||
|
||
async def delete(self, key: str):
|
||
"""删除键"""
|
||
await self.conn.execute("DELETE FROM keyval WHERE key = ?", (key,))
|
||
await self.conn.commit()
|
||
|
||
async def close(self):
|
||
"""关闭连接"""
|
||
if self.conn:
|
||
await self.conn.close()
|