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.2.0-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。

Python SDK 链式 API（0.2.0）

0.2.0 开始，数据库与文档方法均只构造查询，必须调用 .shot() 才发送。0.1.x 的立即执行写法需要在末尾补上 .shot()；conn() 建立连接、login() 登录和后台心跳仍立即执行。

import misakadb as misaka

with misaka.conn(ip="127.0.0.1", port=10032,
                 username="your_user", password="your_password") as conn:
    cur = misaka.getCurrent()
    # 已有数据库和表：保存作用域不会发请求。
    table = cur.use("database01").table("table01")
    table.join({"age": 23, "name": "Lisi"}).shot()
    table.join({"age": 15, "name": "Lisit2", "studentID": 1}).shot()
    print(table.show().magic("+studentID").shot())

    # 成功执行显式作用域后，后续可复用当前 session。
    cur.use().join({"name": "another"}).shot()
    rows = cur.show().magic("+age", "$age>18").limit(100).shot()["data"]
    cur.update().where({"name": "Lisi"}).new({"age": 24}).shot()
    cur.remove().where({"name": "another"}).shot()

示例要求 database01/table01 已存在。cur.use("database01").table("table01").join({...}).shot() 也可一行执行，普通链只发送最后的操作一次。

useOrCreate：使用或创建数据库

# 直到 shot() 才发请求：库存在就使用，不存在就创建后使用。
cur.useOrCreate("example").shot()
# 也可以延续建表链；此例要求 people 尚未创建。
cur.useOrCreate("example").createTable("people").shot()
# 已有表的插入也可链接。
cur.useOrCreate("example").table("people").join({"id": 1, "age": 20}).shot()

useOrCreate() 创建数据库，不自动创建表。沿用当前服务端协议：在 .shot() 内尝试建库，只有明确的“已存在”结果可继续，然后校验并选择数据库，最后执行链尾操作。同名库的并发创建由服务端生命周期锁串行校验；其他创建错误（包括鉴权、权限、I/O 不确定）仍会抛出。

这是多个请求组成的流程，并非跨命令事务：若建库成功后链尾操作失败，新库会保留。完整 .shot() 失败时保留原客户端选择；断线时清空 session。

构造、筛选和结果

• use(database) 选数据库，table(name) 选表；use() 不传参复用构造时 session 的当前数据库和表。构造链不修改 session，显式选库/选表的链在 .shot() 成功后才缓存选择。
• 查询对象不可变，可保存后分支复用。之后切换 session 不会改变已有查询的目标；repr(query)、print(query) 和 query.miql 不触发发送。
• join(document=None) / join().new(document) 构造插入；无参数 join().shot() 插入一个带自动 id 的文档。
• show().where({...}).magic("+age", "$age>18").limit(100).offset(0).shot() 支持连续筛选。多个 where/magic 按 AND 合并，magic("$age") 投影字段值。
• update().where({...}).new({...}).shot() 和 remove().magic("-age").shot() 均可链接。全表修改显式使用 .all()。
• 原来的关键字筛选仍可构造查询，例如 show(where={"id": 1}, magic="+age")、update({"age": 21}, where={"id": 1}) 和 remove(all=True)，末尾均需 .shot()。
• .shot() 返回 Python 字典：插入包含 id 和 affected；查询/更新/删除包含 data、matched、affected、revision、total 和 next_offset。读记录用 result["data"]。
• 每次 .shot() 都执行一次请求，不自动缓存或重试写入。不要重复发送结果不确定的写操作。
• getCurrent() 的连接指针按线程/协程上下文保存；显式共享连接时，共用 session 和请求锁。conn.getCurrent() 始终取得该连接的查询入口。
• 未连接/断线时抛出 ConnectionError，服务端拒绝操作时抛出 misaka.MisakaError，非法参数抛出 ValueError。普通请求的错误只在 .shot() 发生。

table.query().show().where({...}).shot() 继续提供原生 MiQL 构造器，返回原始 bytes；它不接受尚待执行的 useOrCreate 计划。

原有 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，客户端会自动退回为字符串返回