66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
from app.db import DbOper
|
|
from app.db.models.plugin import PluginData
|
|
from app.utils.object import ObjectUtils
|
|
|
|
|
|
class PluginDataOper(DbOper):
|
|
"""
|
|
插件数据管理
|
|
"""
|
|
|
|
def save(self, plugin_id: str, key: str, value: Any) -> PluginData:
|
|
"""
|
|
保存插件数据
|
|
:param plugin_id: 插件id
|
|
:param key: 数据key
|
|
:param value: 数据值
|
|
"""
|
|
if ObjectUtils.is_obj(value):
|
|
value = json.dumps(value)
|
|
plugin = PluginData.get_plugin_data_by_key(self._db, plugin_id, key)
|
|
if plugin:
|
|
plugin.update(self._db, {
|
|
"value": value
|
|
})
|
|
return plugin
|
|
else:
|
|
plugin = PluginData(plugin_id=plugin_id, key=key, value=value)
|
|
return plugin.create(self._db)
|
|
|
|
def get_data(self, plugin_id: str, key: str) -> Any:
|
|
"""
|
|
获取插件数据
|
|
:param plugin_id: 插件id
|
|
:param key: 数据key
|
|
"""
|
|
data = PluginData.get_plugin_data_by_key(self._db, plugin_id, key)
|
|
if not data:
|
|
return None
|
|
if ObjectUtils.is_obj(data.value):
|
|
return json.loads(data.value)
|
|
return data.value
|
|
|
|
def del_data(self, plugin_id: str, key: str) -> Any:
|
|
"""
|
|
删除插件数据
|
|
:param plugin_id: 插件id
|
|
:param key: 数据key
|
|
"""
|
|
PluginData.del_plugin_data_by_key(self._db, plugin_id, key)
|
|
|
|
def truncate(self):
|
|
"""
|
|
清空插件数据
|
|
"""
|
|
PluginData.truncate(self._db)
|
|
|
|
def get_data_all(self, plugin_id: str) -> Any:
|
|
"""
|
|
获取插件所有数据
|
|
:param plugin_id: 插件id
|
|
"""
|
|
return PluginData.get_plugin_data_by_plugin_id(self._db, plugin_id)
|