SQL、NoSQL,以及我为什么开始做 Kanta DB

26 Aug 2026

MongoDB 消除了一些摩擦

MongoDB 让数据的形态更接近我在程序中使用的形态。

我可以直接存储嵌套文档,不用改表就能加字段,还能让新旧记录共存,由应用同时理解两者。这常常让 schema 演进少了很多仪式感。我不必先迁移整个数据库,有时可以在读取或修改旧文档时顺手升级它们。

schema 依然存在。我的代码仍然期望某些字段具有某些含义。MongoDB 只是在何时强制执行这个约定上给了我更多自由。

Redis 给了我出色的部件

Redis 实用得令人赞叹。

如果我需要缓存、队列、计数器、有序集合、分布式锁或临时共享状态,Redis 通常都有紧凑的答案。

当然,当我的数据是 JSON 或其他结构化格式时,要么由我自己把它拆成那些 Redis 原语,要么干脆整体 dump 进去,无视所有那些精巧的工具。

用过以上所有这些,并且总是发现 ORM 在我的代码库里蔓延到无法忍受之后,我终于开始动手实现一个我默默酝酿了很多年的想法。

自己造一个数据库。你知道的,就是那种别人总是劝你连试都别试的东西。

让日志成为数据库

这成了 Kanta 的起点。

我不再想把最新状态作为主要记录、然后在它周围补上历史,我想存储的是变更。

一个对象从空状态开始。之后每个操作只记录改变了什么,连同时间戳和属于这次变更的任何其他元数据。

当前状态来自对日志的应用。现在历史不再需要自己的 schema。查询也不再需要去找最近的时间戳,因为它们作用于任意给定时刻的状态对象。

自始至终一个结构

Msgspec 给我类型化、紧凑的数据结构和非常快的序列化。类似于 dataclasses 或 Pydantic,它能处理嵌套结构和常见的原生类型,比如 UUID、枚举和日期时间,而不会把对象变成 ORM 实体。定义数据结构变得这么简单:

class Data(msgspec.Struct):
    servername: str
    users: dict[UUID, User]   

这意味着我可以在整个程序中使用同一种对象。我可以序列化它。我可以把它发到网络上。我可以持久化它的变更。我可以在另一端重建它。

我不再需要一个类给应用、另一个类给 SQLAlchemy、另一份 schema 给序列化,再加上一堆在它们之间斡旋的小转换函数。

我拼凑了一个简单的 JSONL 记录器,每行一条变更,附上 jsondiff 的变更记录。它不是我可能更喜欢的二进制格式,但它简单到极点,方便编辑和调试。

读取就是从 Python 变量里读,比 Redis 或任何外部数据库都快得多!

写入需要额外的思考。我们不再告诉数据库要改什么,而是直接编辑状态,由数据库持久化一条变更记录。

with kanta.transaction(action="new_user"):
    data.users[uuid7()] = User(...)

注意,尽管我们在写异步 Python,这里却没有 async with 或 await。事务是即时且同步的。你不需要在它周围加任何锁或同步。前后的状态会被保存并比较,从而生成变更 diff。出错时,之前的状态会被恢复,回滚该事务中已经做的一切。

变更由后台线程持久化到磁盘,写入一个 append-only 文件,在断电或崩溃导致损坏时也很容易修复。

差不多就在那时,Kanta 不再像一个数据库实验,而开始像一个可以构建于其上的连贯模型。

在自己的项目上试用之后,我很快把它接入了一个更严肃的应用,它有更重的数据和许多用户。这打消了我对潜在性能问题的担忧,因为毕竟我们实际上是在用 JSON 做数据库,而且用的是一种据我所知自计算时代早期以来就没人用过的日志数据库结构。

果不其然,随着时间推移,应用启动时回放大量变更集变得很慢,于是我加入了完整快照行,避免从初始版本开始长时间回放。在那之后,性能已经超过了我所有的需求。

但重点不是性能,而是简单。只要假设我们的应用可以活在单个 worker 进程里、把完整状态维持在内存中,我们就消除了典型数据库带来的大部分问题。

由于日志结构,回溯历史是免费得来的。我甚至撤销过历史中间的一系列事务——为了拯救一个删掉了部分项目、之后又做了进一步修改的用户。

这个网站也是构建在 Kanta 之上的(Pagerite CMS)。

默认格式刻意保持朴素:JSON 记录,每行一条。Msgspec 把字节和其他类型编码进去。

在发布之前,我想回应一个我自己肯定会提出的批评——缺少二进制支持——于是把二进制实现为一个选项。

用 JSONL 时,我们可以从文件末尾寻找换行符来找到快照,但二进制数据可能会干扰解析器。我们可以从文件开头读取以获得完全确定性的方案,但如果数据库很大,我们更愿意在靠近末尾处寻址。我还想保持 append-only,所以我们不能简单地在文件开头写入快照偏移量。

解决方案是双重的:

• 不可预测的随机 nonce 作为同步字
• Blake3 哈希用于完整性校验

变更负载本身目前用 MessagePack 编码,msgspec 对其有现成支持。

我想把它替换成更接近 Protocol Buffers 的东西,完全不存储字段名和类型,利用 msgspec 已经知道我的结构体的结构和类型这一点,它理应能根据同样的信息推导出紧凑的二进制表示。

这将保持我最看重的性质:数据结构只有一个定义,而不是一个 Python 模型加一个 ORM 模型再加一份序列化 schema。

把权威状态保存在内存里,让另一件事难以忽视:大多数交互式应用本来就在别处保存着另一份副本。

浏览器里也有一份。

一旦数据库本身记录了一个有序的变更流,用同一个流做同步就成了显而易见的下一步。

通过 WebSocket 同步 store

FastAPI 应用通过 WebSocket 发送变更,并把收到的变更应用回数据库。

在每个客户端上,我们维护一个影子副本,即最后一次见到的服务器状态,以及客户端当前的工作状态,存于它原生的 Vue Pinia store。或 Svelte、React 的等价物。

store 是连接的纽带,提供从数据库一路到用户界面再返回的响应性。

客户端可以离线工作,或与其他客户端并发工作,变更到达服务器时被合并,使用带自动解决的三方合并。结合覆盖和验证/拒绝——我们绝对不希望合并冲突需要手动的 git 式解决。

我目前把这套机制嵌在应用里,而不是打包为 Kanta 的一部分。下一步是把它提取成一个独立的 Kanta 兼容同步模块,并为 Javascript 前端提供适配器。

那大概是另一篇文章了。

所有这些设计选择只有在让普通应用代码更简单时才有意义。

用 Kanta,数据库不是我通过另一个模型去查询的东西。我定义想要的状态,打开它,然后操作普通的 Python 对象。读就是读:

user = data.users[user_id]

当我想改点什么,我在一个事务里做出那个变更:

with kanta.transaction(action="update"):
    data.users[user_id].name = "Alice"

中间没有查询语言,没有需要翻译回我的应用模型的 ORM 对象,也没有需要单独记住的审计代码。事务、历史、回滚和持久化全部来自同一个操作。

uv add kanta

用 uv 把它加进你的项目,或者到 git.zi.fi 阅读更多内容。

uv run --with kanta demo.py

import asyncio, msgspec
from kanta import Kanta, configure_logging

# For demonstration purposes, we use "original v0" and "modified v1" in this same script
# Normally your app would only have the latest supported data model

class Data(msgspec.Struct):  # type: ignore - intentionally redefined later
    users: dict[str, dict] = {}
    counter: int = 0

kanta_v0 = Kanta("demo.kantadb", Data())

@kanta_v0.bootstrap
def bootstrap(data: Data) -> None:
    """Create the initial admin user."""
    data.users["userid001"] = {"name": "Alice", "role": "admin"}


# Redefinition to simulate new version
class Data(msgspec.Struct):
    users: dict[str, dict] = {}
    total: int = 0  # Replaces old counter field
    lang: str = "en"  # New field

def migrate_v1(d: dict) -> None:
    """Rename counter to total"""
    d["total"] = d["counter"]

kanta_v1 = Kanta("demo.kantadb", Data(), migrations="demo")

@kanta_v1.logfmt
def resolve_user(value: str, path: str, state: dict) -> str | None:
    """Resolve user ids to names from the database state itself."""
    if path != "$user" and not path.startswith("users."):
        return None
    return state.get("users", {}).get(value, {}).get("name")


async def main() -> None:
    print("Database creation with v0 schema and basic transactions:\n")
    # Open and close automatically; you can also `await kanta.open()` instead
    async with kanta_v0 as kanta:
        with kanta.transaction(action="create", user="userid001") as data:
            data.users["userid002"] = {"name": "Bob", "role": "user"}

        with kanta.transaction(action="update", user="userid001") as data:
            data.users["userid002"]["role"] = "editor"
            data.counter = 1

        # Display-only extra string, appended after the action.
        with kanta.transaction(action="export", user="userid002", extra="(we are still v0)") as data:
            data.counter = 2

    print("\nA new data model, migrations and logfmt pretty names:\n")
    async with kanta_v1 as kanta:
        with kanta.transaction(action="update", user="userid002", extra="(new version)") as data:
            data.total += 1

        try:
            with kanta.transaction(action="reset", user="userid001") as data:
                data.total = 99
                raise ValueError("simulated failure")
        except ValueError:
            print(f"\nReset rolled back: {data.total=} (we can always read data without tx)\n")

        with kanta.transaction(action="delete", user="userid002") as data:
            data.users.pop("userid001", None)  # del if exists

if __name__ == "__main__":
    configure_logging(debug=True)
    asyncio.run(main())