SQL、NoSQL，以及我为何创立了 Kanta DB

SQL 之所以仍然是默认选项，是有其深层原因的……

如果我今天要搭建一个传统的Python服务，我很可能会从PostgreSQL、SQLAlchemy和Alembic入手。PostgreSQL能提供真正的UUID、JSONB、数组、事务、约束和出色的索引功能。SQLAlchemy能将这些功能大多无缝地映射到Python中。Alembic则能清晰地记录和版本化数据库架构的变更。

通常情况下，在任何事情开始让我感到厌烦之前，我都能坚持很久。

…但这个模型有边缘部分

数据库模式和 Python 数据结构并不是完全相同的概念。

借助 SQLAlchemy，我可以定义类型化模型，直接对 PostgreSQL 的 UUID 列进行操作，将 JSONB 映射到 Python 容器中，并将大多数常规转换操作从我自己的代码中移除。这比将每个数据库值都当作字符串处理，或手动拼凑 SQL 语句要好得多 uuid.UUID 使用 SQLAlchemy，我可以定义类型化模型，直接操作 PostgreSQL 的 UUID 列，将 JSONB 映射到 Python 容器，并将大多数常规转换操作从我自己的代码中移除。这比将每个数据库值都视为字符串或手动拼凑 SQL 语句要好得多。

尽管如此，ORM模型还是会成为一种特殊的对象类型。它承载着数据库表的列、表之间的关联关系、会话行为以及持久化规则。如果程序的其他部分需要更简洁的应用架构，我要么将数据库相关的逻辑分散到其他模块中，要么添加一层额外的转换层来处理这些逻辑。

当数据模型发生变化时，这个问题会变得更加明显。

向Python结构中添加一个字段看似微不足道。但要在持久化数据中添加一列，就意味着要更改模型并创建迁移脚本。更复杂的改动则需要进行数据转换和兼容性评估。Alembic能妥善处理这些问题，但我仍需维护结构变更的历史记录，以便旧行能转换为新行。

这并非PostgreSQL的缺陷。该数据库已采用了固定的架构，因此对其进行更改会产生相应的影响。

历史数据会形成另一层信息。如果我想知道是谁更改了某个值、更改的时间，或者上周二的记录是什么样的，我需要对此进行建模。我可以添加审计表、触发器、时间戳列或事件溯源层。PostgreSQL 可以很好地支持所有这些功能。

但现实中的生活依然是当下真实存在的样子。而历史，则是我围绕它所构建的虚构叙事。

SQL 的替代方案

来一杯美味的 NoSQL 怎么样

MongoDB 消除了某些操作上的阻碍

MongoDB 将数据的结构调整得与我程序中使用的结构更为契合。

我可以直接存储嵌套文档，无需修改表格即可添加字段，并且让旧记录和新记录共存，同时应用程序能兼容两者。这样一来，数据库模式的演进往往变得不那么繁琐了。我无需先迁移整个数据库，有时在读取或修改旧文档时，就能直接对其进行升级了。

数据模型仍然存在。我的代码仍然期望某些字段具有特定的含义。MongoDB只是让我能更灵活地选择何时执行这项约束条件。

Redis 总是能给我提供一些绝佳的素材

Redis 实用性极强。

如果我需要缓存、队列、计数器、排序集、分布式锁或临时共享状态，Redis通常都能提供一个简洁高效的解决方案。

当然，如果我的数据是以 JSON 或其他结构化格式呈现的，那么将其拆解成 Redis 原生数据类型，或者直接以完整形式导入而无需借助任何高级工具，这完全取决于我的个人选择。

推拉

要是有人动了我的数据，记得通知我一声

Firebase 的核心功能始于数据同步

Firebase则采取了一种更为直接的做法。其数据库将实时客户端更新视为产品的一部分。

我可以为数据添加监听器，让客户端在数据发生变化时即时接收更新。离线行为和重新连接功能也属于同一套系统，而非作为单独的 WebSocket 项目在后期才添加进来。

那真是太吸引人了。

所有变动我们都已经处理好了

MongoDB对此有强大的内置解决方案。变更流（Change Streams）让我能够监视一个集合、数据库或整个部署，并接收插入、更新、删除等操作。更新通常会包含已更改的字段，每个事件都带有恢复令牌，因此只要操作日志（oplog）中仍包含该点，用户就可以重新连接并从中断的地方继续处理。

PostgreSQL 还能够通过逻辑解码和复制功能，展示数据库的实际变更情况。在这方面，Redis 的变更流（Change Streams）也提供了类似的解决方案。

变更数据捕获系统在此基础上构建了功能强大的数据管道，但随后我们需要解析 SQL 语句或 Redis 命令，并自行跟踪其状态变化。

发布和订阅

PostgreSQL 和 Redis 也都提供了传统的消息通道功能。

我可以使用 LISTEN/NOTIFY 或 PUB/SUB 机制，并在同一事务中发送通知来修改数据。这样一来，就避免了使用无关的消息总线所带来的种种麻烦了。

但我仍然需要创建和接收通知消息，以决定阅读什么内容和发送什么信息。此外，在实际更改与通知之间还存在竞争条件的问题。

就我遇到的这个问题而言，感觉我起点太低了，远远达不到我真正想要的抽象层级。

我不只是想知道那个状态发生了变化。
我想要的是变化本身，以及变化前后的状态。

或许我该自己动手做一个？

用过上述所有工具后，我发现ORM在代码库中的蔓延愈发难以忍受。最终，我开始着手实现一个我默默构思多年的想法。

去创建自己的数据库。你懂的，这种事他们总是叫你别去尝试的。

让日志文件充当数据库的作用

这成为了康塔行动的起点。

我不想把最新的状态作为主要记录并在其周围添加历史记录，而是想存储所有的变更记录。

一个对象最初处于空状态。后续的每一次操作只会记录发生的变化，以及对应的时戳和与该变化相关的其他元数据。

当前状态是通过应用日志生成的。现在，历史数据不再需要单独的数据模型。查询也不必查找最新的时间戳，因为它们可以在任意时刻直接访问状态对象进行处理。

全程只需搭建一个结构即可

Msgspec gives me typed, compact data structures with very fast serialization. Similar to dataclasses or Pydantic, it handles nested structures and common native types such as UUIDs, enums and datetimes without turning the objects into ORM entities. Defining your data structures becomes this simple:

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

这意味着我在整个程序中可以使用同一种对象。我可以对其进行序列化处理，可以通过网络传输它，还能保存其修改内容，并且可以在另一端成功地将其还原。

我的应用程序不需要一个类来处理数据库连接，另一个类用于 SQLAlchemy，第三个类用于序列化，还有一些转换函数在它们之间进行数据转换。

它还活着呢

我临时拼凑了一个简单的 JSONL 日志记录器，每行一个更改，并附带一个 jsondiff 更改记录。虽然它不是我理想中的二进制格式，但编辑和调试起来却极其简单方便。

读取操作只需从 Python 变量中读取数据，速度比 Redis 或任何外部数据库都要快得多！

写入操作需要额外的思考。与其告诉数据库该更改什么内容，不如先编辑数据状态，之后由数据库自动记录更改日志。

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

请注意，尽管我们正在开发异步Python，但这里并没有 或 相关的操作。交易是即时且同步的。无需对其进行任何锁定或同步操作。交易前后的状态会被存储并进行比较，以生成变更差异。若出现错误，系统会恢复到之前的状态，回滚该交易中已完成的所有操作 async with 或者 await 尽管我们正在开发异步 Python 功能，但这里所涉及的交易是即时且同步的。无需对其进行任何锁定或同步操作。交易前后的状态会被存储并进行比较，以生成变更差异。若出现错误，系统会恢复到之前的状态，回滚该交易中已完成的所有操作。

这些更改由后台线程写入磁盘，存储在一个只读写文件中。即使因电源故障或系统崩溃导致数据损坏，该文件也可轻松修复。

差不多就在那时，Kanta 不再像是一个数据库实验品，而开始呈现出一个可以进一步完善的连贯模型雏形。

何不把它应用到生产环境中试试看呢？

在对自己的项目进行测试后，我很快将其应用于一个数据量更大、用户数量众多的严肃应用程序中。这消除了我对可能出现性能问题的担忧，毕竟我们实际上是在用 JSON 来模拟数据库，而这种日志数据库结构据我所知，自计算技术诞生之初就鲜有其他人使用过。

果然，应用启动时重放大量变更集会变得很慢。于是我添加了完整快照行，以避免从初始修订版开始进行长时间的重放。自此之后，应用的性能完全满足了我的需求。

但关键点不在于性能，而在于简洁性。假设我们的应用可以运行在单个工作进程中，并在其内存中维护完整的状态，这样一来，我们就能规避掉传统数据库所带来的大部分问题。

由于日志结构的原因，回滚历史操作是免费的。我甚至曾在历史记录中回滚过一系列事务——只为挽救一位用户，他先是删除了项目的部分内容，随后又对其进行了进一步的修改。

该网站也是基于Kanta（Pagerite内容管理系统）搭建的。

迁徙

图片
检查数据库中的任意切片，使用正则表达式搜索你需要的数据

模具加工

Kanta also has a small CLI for inspecting databases directly. It can replay a database, inspect selected ranges of its history, load the application’s actual data types and run migrations when needed, and dump the resulting state as JSON. A good tool also beats treating it as a text file.

单进程模型省去了大量硬件设备，但并不能消除软件会随时间变化这一事实。

没人会喜欢迁移操作。迁移数据意味着要承受维护更改的重负。新增一个 SQL 迁移任务，或是另设一个 Mongo 遗留备用分支和更新分支，实在是件麻烦事，所以人们往往宁愿避免进行这些改动。

再次强调，msgspec 承担了大部分繁重工作。若想添加或删除某个字段，只需将新字段及其默认值添加到数据结构中，或删除任何旧字段即可。系统会自动将其迁移至新格式。若加载的格式与数据结构不匹配，系统会提示错误，指出具体问题所在及出错位置。

很酷也很简单，但对于一个数据库来说还不够。

时不时地，我们会想要重命名某个字段、更改数据类型，或者彻底重组整个数据结构，甚至可能在此过程中获取新的外部数据（我自己就这么做过）。这些操作都需要一个真正懂行、知道自己在做什么的迁移函数来完成。

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

格式很简单：数据库版本号直接来自函数名称。该函数处理纯字典格式，因此我们无需维护旧版本的msgspec.Structs。文档字符串会提供已记录的操作描述。

為了避免在我们的程序中到处乱添加这些代码，我们可以将迁移功能放在一个单独的 Python 模块中。在定义数据库时，只需使用 Python 模块的路径来引用该模块即可：

kanta = Kanta("foo.kantadb", migrations="foo.migrations")

变更集元数据和快照包含它们所代表的版本号。我们会从该版本向上应用所有找到的迁移函数，并无论如何都要完成消息规范（msgspec）转换。如果有任何相关变更发生，我们会记录并存储已完成的迁移操作，并在此后生成快照。

若某些版本不再需要支持，最古老的迁移功能也可以被移除，这样一来，整个迁移系统就能保持易于管理了。

关于磁盘格式

默认格式特意设计得比较简洁：每行一个 JSON 记录。Msgspec 会将字节和其他类型的数据编码成这种格式。

在发布之前，我想先回应一下我自己可能会提出的批评意见，即该产品缺乏对二进制文件的支持。为此，我决定将二进制支持功能作为一项可选项来实现。

使用 JSONL 时，我们可以从文件末尾向前搜索换行符来查找快照，但二进制数据可能会让解析器感到困惑。为了实现完全确定性的处理方式，我们可以从文件开头读取数据，但如果数据库规模较大，我们更倾向于从接近文件末尾的位置开始搜索。此外，我还希望保持数据的只增不减特性，因此我们不能简单地将快照偏移量写入文件开头。

解决方案分为两个方面：

  • 使用不可预测的随机随机数作为同步字（syncword）
  • 用于完整性验证的 Blake3 哈希值

这些更改数据包本身目前使用 MessagePack 进行编码，而 MessagePack 已在 msgspec 中得到了广泛支持。

我想用更接近 Protocol Buffers 的方案来替代它，完全不存储字段名称和类型。利用 msgspec 已经了解我的结构体的结构和类型这一优势，它应该能够从这些信息中生成一个紧凑的二进制表示形式。

这样一来，我最关心的特性就能得以保留：只需定义一次数据结构，而无需分别维护 Python 模型、ORM 模型以及序列化方案。

数据库只是整个系统的一半功能而已

将权威状态保存在内存中会引发另一个不容忽视的问题：大多数交互式应用程序通常会将另一份副本存储在其他位置。

浏览器也有这个功能。

既然数据库本身会记录一系列有序的变更，那么利用这套变更记录来实现数据同步，就成了顺理成章的下一步举措。

通过WebSocket同步商店数据

FastAPI应用程序通过WebSocket发送更改，并将收到的更改应用回数据库中。

针对每位客户端用户，我们会维护一份影子副本，其中包含服务器的最后可见状态，以及客户端在其原生 Vue Pinia 存储中当前的工作状态。Svelte 或 React 框架的用户也可使用类似的功能。

商店是整个链条中的关键环节，它能实现从数据库到用户界面，再从用户界面到数据库的全程交互响应。

客户端可以离线工作，也可与其他客户端并行协作。当更改内容提交至服务器时，系统会采用三方合并机制并自动解决冲突。结合覆盖写入功能及验证/拒绝机制，我们绝不希望合并冲突需要手动采用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](git 了解更多信息。zi.fi/LeoVasanko/kanta).

图片
演示部分特意超出了”Hello World”的范畴。它先创建一个数据库，使用一个版本的数据模型进行初始化，然后通过多笔事务对数据库进行修改。接着，再用更新后的数据模型打开同一数据库，运行迁移操作，演示回滚功能，最后继续正常使用该数据库。大致来说，这就是我想要让其显得平淡无奇的整个生命周期流程。
uv run --with kanta demo.py

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())