SQLite 多租户示例：同号任务的复合主键与请求结果隔离

import json
import sqlite3
from contextlib import closing
from dataclasses import dataclass


@dataclass(frozen=True)
class Session:
    tenant: str
    actor: str
    can_claim: bool


def read_task(db, session, task_id):
    if not session.can_claim:
        raise PermissionError("permission revoked")
    return db.execute(
        "SELECT state, version FROM task WHERE tenant=? AND id=?",
        (session.tenant, task_id)
    ).fetchone()


def claim(db, session, task_id, request_id, expected):
    if not session.can_claim:
        raise PermissionError("permission revoked")
    payload = json.dumps([task_id, expected], separators=(",", ":"))
    scope = (session.tenant, session.actor, request_id)
    with db:
        prior = db.execute(
            "SELECT payload, result FROM operation "
            "WHERE tenant=? AND actor=? AND request_id=?", scope
        ).fetchone()
        if prior:
            if prior[0] != payload:
                raise ValueError("request payload conflict")
            return prior[1]
        changed = db.execute(
            "UPDATE task SET state='claimed', version=version+1 "
            "WHERE tenant=? AND id=? AND state='ready' AND version=?",
            (session.tenant, task_id, expected)
        ).rowcount
        if changed != 1:
            return "conflict"
        db.execute("INSERT INTO operation VALUES (?, ?, ?, ?, ?)",
                   (*scope, payload, "accepted"))
        return "accepted"


def main():
    with closing(sqlite3.connect(":memory:")) as db:
        db.executescript("""
        CREATE TABLE task (tenant TEXT, id INTEGER, state TEXT, version INTEGER,
          PRIMARY KEY(tenant, id));
        CREATE TABLE operation (tenant TEXT, actor TEXT, request_id TEXT,
          payload TEXT, result TEXT, PRIMARY KEY(tenant, actor, request_id));
        INSERT INTO task VALUES ('A', 1, 'ready', 1);
        INSERT INTO task VALUES ('B', 1, 'ready', 1);
        """)
        a = Session("A", "rider", True)
        b = Session("B", "rider", True)
        assert len(db.execute("SELECT tenant FROM task WHERE id=1").fetchall()) == 2
        assert claim(db, a, 1, "req-1", 1) == "accepted"
        assert read_task(db, a, 1) == ("claimed", 2)
        assert read_task(db, b, 1) == ("ready", 1)
        assert claim(db, a, 1, "req-1", 1) == "accepted"
        assert claim(db, a, 1, "new-key", 1) == "conflict"
        assert claim(db, b, 1, "req-1", 1) == "accepted"
        try:
            claim(db, a, 2, "req-1", 1)
        except ValueError:
            pass
        else:
            raise AssertionError("key reused for another task")
        try:
            claim(db, Session("A", "rider", False), 1, "req-1", 1)
        except PermissionError:
            pass
        else:
            raise AssertionError("revoked session read cached result")
        assert db.execute("SELECT COUNT(*) FROM operation").fetchone()[0] == 2
        print("unscoped task lookup: 2 rows")
        print("A claim: B still ready")
        print("same request key: independent tenant results")
        print("stale version: conflict")
        print("changed payload and revoked permission: rejected")
        print("operation records: 2")


if __name__ == "__main__":
    main()

python demo.py

unscoped task lookup: 2 rows
A claim: B still ready
same request key: independent tenant results
stale version: conflict
changed payload and revoked permission: rejected
operation records: 2