MisakaDB Python Client

作者：刘博源 这是 MisakaDB 的 Python 客户端文档，主要面向客户端调用方。

包名和导入名均为 misakadb，需要 Python 3.11 或更高版本。直接入口为 conn() 和 getCurrent()；同时保留 MisakaDBClient、MiQL 等原有接口。客户端只依赖 Python 标准库。

功能概览

• 连接 MisakaDB 服务端
• 提供身份验证 (Login) 与安全拦截
• 获取服务信息与动态面板展示
• 执行文本命令
• 查询服务版本、网络配置、允许命令
• 自动发送心跳包
• 手动启动、停止心跳
• 查询心跳次数、失败次数、丢包率

目录

• 客户端 API: misakadb/apis/api.py
• 命令行入口: misakadb/cli.py
• 底层 socket 实现: misakadb/network/sock.py
• 命令格式化输出: misakadb/network/command_send.py
• 压测示例: usage/demo1.py

构建与安装

从 PyPI 安装客户端：

python -m pip install misakadb

在仓库根目录安装源码包：

python -m pip install ./client

构建 wheel 和源码分发包（构建工具使用仓库已有的 uv）：

uv build client --out-dir client/dist/packages
python -m pip install client/dist/packages/misakadb-0.1.2-py3-none-any.whl

已有 wheel 时可直接拷贝到其他机器，用 python -m pip install <wheel路径> 安装，无需访问包索引。发布目标为 PyPI 的 misakadb 项目，公开发布成功后可直接通过包名安装。

安装后可运行：

misakadb --version
misakadb --address 127.0.0.1 --port 10032 --mode shell
# 等价入口
python -m misakadb --address 127.0.0.1 --port 10032 --mode shell

使用虚拟环境时先激活环境；项目现有环境可用 source client/.venv/bin/activate。

直接使用 API

import misakadb as misaka

conn = misaka.conn(ip="127.0.0.1", port=10032)
# 如果服务端启用了鉴权：
conn.login("your_user", "your_password")
mq = misaka.getCurrent()

# 数据库 test 和表 people 需要已经存在。
result = mq.use("test").table("people").join({"id": 1, "name": "御坂", "age": 20})
print(result)  # {"id": 1, "affected": 1}
conn.close()

conn() 立即建立连接，也可以通过 username=...、password=... 在连接后自动登录。它返回连接对象并设置当前上下文的连接。无需创建 MiQL，无需调用 .shot()，操作执行后直接返回 Python 字典。join() 不传数据时会插入一个只有自动生成 id 的文档。

创建数据库、创建表及完整 CRUD：

import misakadb as misaka

with misaka.conn(ip="127.0.0.1", port=10032,
                 username="your_user", password="your_password") as conn:
    mq = misaka.getCurrent()
    mq.createDB("example")
    db = mq.use("example")
    db.createTable("people")
    table = db.table("people")

    table.join({"id": 1, "age": 20})
    rows = table.show(where={"id": 1})["data"]
    table.show(magic=["+age", "$age>18"], limit=100, offset=0)
    table.update({"age": 21}, where={"id": 1})
    table.remove(where={"id": 1})

• mq.use(database) 选择数据库并返回数据库作用域，.table(name) 选择表并返回表作用域。两者仅在服务端确认成功后更新该连接的 session。
• 选择后可以直接调用 mq.join(...)、mq.show(...)、mq.update(...)、mq.remove(...)，使用 session 的当前表。
• 保存的 table 对象绑定原来的连接、数据库和表。之后切换当前选择不会改变该对象的目标，重复操作无需重复执行 use/table。
• show(where=None, *, magic=None, limit=None, offset=0) 立即查询。magic 可以是单个字符串或字符串列表，where 与 magic 同时提供时取 AND。
• update(document, *, where=None, magic=None, all=False) 立即按补丁更新；remove(where=None, *, magic=None, all=False) 立即删除。全表修改必须显式使用 all=True。
• join() 返回 id 和 affected；其他 CRUD 返回包含 data、matched、affected、revision、total 和 next_offset 的字典，show()["data"] 取得记录列表。
• getCurrent() 的当前连接指针按线程/协程上下文保存；新线程应单独调用 conn()。继承或显式共享同一连接时，共用该连接的 session 和请求锁。已保存的 conn.getCurrent() 可随时取得该连接的 API。
• 未连接或连接已关闭时抛出 ConnectionError，服务端拒绝操作时抛出 misaka.MisakaError，非法 API 参数抛出 ValueError。新连接或登录失败不会替换已有的当前连接。

需要原来的延迟构造语法时，可以用 table.query().show().where({"id": 1}).shot()；它仍返回原始 bytes。

原有 MiQL 构造接口

from misakadb import MisakaDBClient, MiQL, commandSend

with MisakaDBClient("127.0.0.1", 10032) as client:
    # 使用服务端已有的用户名与密码。
    if not commandSend(client.client).login("your_user", "your_password"):
        raise PermissionError("登录失败")
    mq = MiQL(client.client)
    print(mq.useDB("test").shot().decode())
    print(mq.createTable("people").shot().decode())
    print(mq.join({"id": 1, "name": "御坂", "age": 20}).shot().decode())
    print(mq.show().magic("$age>18").shot().decode())
    print(mq.update().where({"id": 1}).new({"age": 21}).shot().decode())
    print(mq.remove().where({"id": 1}).shot().decode())

示例要求数据库 test 已存在且表 people 尚未创建。已有表可用 mq.use("people").shot() 选择。原生 Python SDK 的字典键需要引号；Shell 字符串解析器允许不带引号的键。查询返回值保持现有的 bytes 协议格式，.shot() 执行请求，成功后的数据库和表选择保存在该连接的 session 中。

仓库中原有的 from apis.api import MisakaDBClient、from mql.MQ import MiQL 和 python client/misaka-cli.py 继续通过兼容入口使用同一份实现。安装包本身只提供 misakadb 命名空间。

快速开始

1. 创建客户端

from misakadb import MisakaDBClient

client = MisakaDBClient("127.0.0.1", 10032)

默认参数：

• host="127.0.0.1"
• port=10032
• heartbeat_interval=1.0

2. 连接服务端与鉴权

ok = client.connect()
if not ok:
    print("连接失败")

# 如果服务端开启了鉴权，需要发送 login 命令
# 在交互模式下，你可以使用 command_send 模块：
# command_send.login(username, password)

也可以指定重试参数：

ok = client.connect(retries=3, retry_delay=1.0)

3. 获取服务信息

service_info = client.get_service_info()
print(service_info)

典型返回值：

{
    "network": {
        "port": 10032,
        "address": "0.0.0.0",
        "max_conn": 2,
        "retry_count": 3,
        "retry_delay": 30
    }
}

4. 关闭连接

client.close()

推荐写法

推荐使用上下文管理器，避免忘记关闭连接：

from misakadb import MisakaDBClient

with MisakaDBClient("127.0.0.1", 10032) as client:
    info = client.get_service_info()
    print(info)

API 说明

MisakaDBClient(host="127.0.0.1", port=10032, heartbeat_interval=1.0)

创建客户端对象。

参数说明：

• host: 服务端地址
• port: 服务端端口
• heartbeat_interval: 心跳发送间隔，单位秒

注意：

• 初始化时会创建 client.heart 心跳控制器
• 初始化时会启动心跳线程
• 真正发送心跳前仍要求连接已建立

connect(retries=3, retry_delay=1.0) -> bool

连接到服务端。

参数说明：

• retries: 失败重试次数
• retry_delay: 重试间隔，单位秒

返回值：

• True: 连接成功
• False: 连接失败

注：此方法仅负责底层 TCP 连接。如果需要身份验证，请在连接成功后调用 execute_command("login <user> <pass>") 或使用 commandSend.login()。

close() -> None

关闭当前连接，并停止当前心跳控制器。

get_service_info() -> dict | None

获取服务端信息。

成功时返回字典，失败时返回 None。

execute_command(command: str) -> str | dict | list | None

发送一条文本命令到服务端。

行为说明：

• 如果返回内容是 JSON，会自动解析成 dict 或 list
• 如果返回内容不是 JSON，会返回字符串
• 如果请求失败，返回 None

示例：

result = client.execute_command("get-service-info")
print(result)

ping() -> bool

用于检查当前服务是否可达。

返回值：

• True: 当前请求成功
• False: 当前请求失败

注意：

• 这里是业务层命令检测
• 不是底层 TCP keepalive
• 也不是服务端心跳回包确认

get_server_version() -> str | None

获取服务端版本号。

get_allowed_commands() -> list[str] | None

获取服务端允许的命令列表。

is_command_allowed(command: str) -> bool

检查某条命令是否被服务端允许。

get_network_config() -> dict | None

获取服务端网络配置。

心跳控制

客户端暴露了一个心跳控制对象：

client.heart

自动行为

• MisakaDBClient 初始化时会启动心跳线程
• connect() 时会确保心跳处于启动状态
• 已连接状态下，心跳线程会定时发送心跳帧

手动启动和停止

client.heart.start()
client.heart.stop()

说明：

• start() 启动或恢复心跳发送
• stop() 停止发送心跳，但不会关闭连接

心跳统计字段

client.heart.count
client.heart.success_count
client.heart.failure_count
client.heart.loss_rate
client.heart.running
client.heart.last_error
client.heart.last_sent_at

字段说明：

• count: 心跳发送尝试总次数
• success_count: 心跳发送成功次数
• failure_count: 心跳发送失败次数
• loss_rate: 当前统计口径下的失败率，单位百分比
• running: 当前是否处于启用状态
• last_error: 最近一次失败错误
• last_sent_at: 最近一次成功发送心跳的时间戳

获取完整统计

stats = client.heart.stats()
print(stats)

返回示例：

{
    "running": True,
    "interval": 10.0,
    "count": 12,
    "success_count": 12,
    "failure_count": 0,
    "loss_rate": 0.0,
    "last_error": None,
    "last_sent_at": 1710000000.123
}

关于“丢包率”的说明

当前协议没有单独的心跳响应包，因此这里的“丢包率”并不是网络层真实丢包率，而是：

心跳发送失败次数 / 心跳发送尝试总次数

如果后续服务端增加了心跳应答包，这里的统计口径可以再升级为真正的收发确认率。

命令行交互模式 (CLI)

客户端提供了一个功能丰富的交互式终端入口 (misakadb 命令)，支持以下特性：

• 密码回显关闭：安全的密码输入体验（与 Linux 终端一致）
• 华丽面板：针对 get-service-info 等特殊命令提供了对齐的 JSON 展平和 Misaka ASCII Logo 渲染
• 自动耗时统计：所有命令自动显示执行延迟 (绿色 < 0.2ms，黄色 >= 0.2ms)
• 多行输入：支持以 \ 结尾进行多行命令拼接

启动 CLI

misakadb --address 127.0.0.1 --port 10032 --mode shell

可选参数：

• --username <name>: 预设用户名
• --password <pass>: 预设密码
• --mode onlyConn: 仅建立连接测试，跳过交互式终端 (适合作为 Windows 后台服务启动)

完整示例

from misakadb import MisakaDBClient

with MisakaDBClient("127.0.0.1", 10032, heartbeat_interval=5.0) as client:
    info = client.get_service_info()
    print("服务信息:", info)

    print("服务版本:", client.get_server_version())
    print("网络配置:", client.get_network_config())
    print("允许命令:", client.get_allowed_commands())

    print("心跳是否运行:", client.heart.running)
    print("心跳次数:", client.heart.count)
    print("心跳失败率:", client.heart.loss_rate)

    client.heart.stop()
    print("手动停止心跳:", client.heart.running)

    client.heart.start()
    print("重新启动心跳:", client.heart.running)

    result = client.execute_command("get-service-info")
    print("命令结果:", result)

注意事项

• execute_command() 和心跳发送共用同一个 socket，并通过锁保证不会交叉写入
• 当前客户端是长连接模型，建议复用连接，不要高频短连接压测
• 如果你要做压测，优先参考 usage/demo1.py 的“单连接多请求”模式
• 如果服务端返回的内容不是合法 JSON，客户端会自动退回为字符串返回