#Context
通过 plugin() 工厂函数获取插件上下文,统一管理持久化、日志和目录。
#PluginContext
数据结构
_AppContext
from dataclasses import dataclass
from pathlib import Path
import logging
@dataclass
class PluginContext:
"""插件上下文"""
name: str # 插件名称
home: Path # 插件专属目录 plugins/{name}/
logger: logging.Logger # 插件专属日志器
app: _AppContext # 应用上下文
def state(self, name: str = "state") -> StateStore:
"""获取插件专属持久化实例"""@dataclass
class _AppContext:
"""应用上下文,跨插件共享"""
logger: logging.Logger # 全局日志器
def store(self, name: str = "plugins.state") -> StateStore:
"""获取全局持久化实例"""#plugin
工厂函数,创建插件上下文。
函数签名
使用示例
def plugin(name: str) -> PluginContext:
"""
Args:
name: 插件名
Returns:
PluginContext,home 为 plugins/{name}/,
state() 默认路径为 plugins/{name}/state.json,
app.store() 默认路径为 state/plugins.state.json,
logger 写入 plugins/{name}/{name}.log
"""from byksdk import plugin
ctx = plugin("server")
# 插件专属持久化
ctx.state().set("port", 8080)
ctx.state("config").set("timeout", 30)
# 插件专属日志
ctx.logger.info("server started")
# 全局持久化
ctx.app.store().set("theme", "dark")
ctx.app.store("shared").set("key", "value")
# 全局日志
ctx.app.logger.info("app started")
# 插件目录
print(ctx.home) # ~/.byk/plugins/server/#持久化路径
~/.byk
plugins
server
state.json# ctx.state()
config.json# ctx.state("config")
server.log# ctx.logger
state
plugins.state.json# ctx.app.store()
shared.json# ctx.app.store("shared")
logs
plugins.log# ctx.app.logger
#StateStore
基于 JSON 文件的持久化存储,通过 plugin() 上下文或直接创建使用。
#创建实例
通过插件上下文创建
直接创建
from byksdk import plugin
ctx = plugin("server")
ctx.state() # plugins/server/state.json
ctx.state("config") # plugins/server/config.json
ctx.app.store() # state/plugins.state.json
ctx.app.store("shared") # state/shared.jsonfrom byksdk import StateStore, PLUGINS_DIR, STATE_DIR
store = StateStore(PLUGINS_DIR / "server" / "state.json")
store = StateStore(STATE_DIR / "shared.json")#API
接口定义
使用示例
class StateStore:
path: Path # 数据文件路径
def load(self) -> dict # 读取全部数据
def save(self, data) -> dict # 覆盖保存,返回保存后的数据
def get(self, key, default) # 获取单个值,key 不存在返回 default
def set(self, key, value) # 设置单个值,返回完整数据
def update(self, values) # 批量更新,返回更新后的数据
def delete(self, key) # 删除单个值,返回剩余数据
def clear(self) # 清空所有数据,返回空字典ctx = plugin("server")
# 写入
ctx.state().set("port", 8080)
ctx.state().set("host", "localhost")
ctx.state().update({"debug": True, "timeout": 30})
# 读取
ctx.state().get("port") # 8080
ctx.state().get("missing", 30) # 30
ctx.state().load() # {'port': 8080, ...}
# 删除
ctx.state().delete("host")
# 覆盖
ctx.state().save({"new": "data"})
# 清空
ctx.state().clear()#注意事项
数据必须是 JSON 可序列化的类型:
# ✅ 支持
ctx.state().set("string", "text")
ctx.state().set("number", 123)
ctx.state().set("list", [1, 2, 3])
ctx.state().set("dict", {"key": "value"})
ctx.state().set("none", None)
# ❌ 不支持
ctx.state().set("set", {1, 2, 3}) # set 不可序列化
ctx.state().set("bytes", b"data") # bytes 不可序列化