Python 批量交接示例：任务改组后，旧批次如何逐项返回冲突

import json
import sqlite3
from contextlib import closing


def freeze_batch(db, batch):
    return db.execute(
        "SELECT id, version FROM task WHERE batch=? ORDER BY id", (batch,)
    ).fetchall()


def move(db, task_id, expected, new_batch):
    with db:
        return db.execute(
            "UPDATE task SET batch=?, version=version+1 "
            "WHERE id=? AND version=? AND state='ready'",
            (new_batch, task_id, expected)
        ).rowcount == 1


def claim_one(db, request_id, batch, actor, task_id, expected):
    # actor is supplied by a trusted session in a real service.
    if actor not in {101, 202}:
        raise PermissionError("actor cannot claim")
    payload = json.dumps([batch, actor, expected], separators=(",", ":"))
    with db:
        prior = db.execute(
            "SELECT payload, result FROM operation WHERE request_id=? AND task_id=?",
            (request_id, task_id)
        ).fetchone()
        if prior:
            if prior[0] != payload:
                raise ValueError("same operation key with another payload")
            return prior[1]
        changed = db.execute(
            "UPDATE task SET state='claimed', owner=?, version=version+1 "
            "WHERE id=? AND batch=? AND version=? AND state='ready'",
            (actor, task_id, batch, expected)
        ).rowcount
        result = "accepted" if changed else "conflict"
        db.execute("INSERT INTO operation VALUES (?, ?, ?, ?)",
                   (request_id, task_id, payload, result))
        return result


def claim_batch(db, request_id, batch, actor, snapshot):
    return [(task_id, claim_one(db, request_id, batch, actor, task_id, version))
            for task_id, version in snapshot]


def main():
    with closing(sqlite3.connect(":memory:")) as db:
        db.executescript("""
        CREATE TABLE task (id INTEGER PRIMARY KEY, batch TEXT,
          version INTEGER, state TEXT, owner INTEGER);
        CREATE TABLE operation (request_id TEXT, task_id INTEGER,
          payload TEXT, result TEXT, PRIMARY KEY(request_id, task_id));
        INSERT INTO task VALUES (1, 'A', 1, 'ready', NULL);
        INSERT INTO task VALUES (2, 'A', 1, 'ready', NULL);
        """)
        snapshot_a = freeze_batch(db, "A")
        assert move(db, 1, 1, "B")
        first = claim_batch(db, "req-A", "A", 101, snapshot_a)
        assert first == [(1, "conflict"), (2, "accepted")]
        assert claim_batch(db, "req-A", "A", 101, snapshot_a) == first
        assert db.execute("SELECT COUNT(*) FROM task").fetchone()[0] == 2
        assert db.execute("SELECT batch, owner FROM task WHERE id=1").fetchone() == ("B", None)
        snapshot_b = freeze_batch(db, "B")
        assert claim_batch(db, "req-B", "B", 202, snapshot_b) == [(1, "accepted")]
        assert not move(db, 2, 2, "C")
        try:
            claim_batch(db, "req-A", "A", 202, snapshot_a)
        except ValueError:
            pass
        else:
            raise AssertionError("payload conflict was accepted")
        print("batch A:", first)
        print("replay: identical item results")
        print("task count after regroup: 2")
        print("batch B: task 1 accepted by 202")
        print("claimed task regroup: rejected")


if __name__ == "__main__":
    main()

python demo.py

batch A: [(1, 'conflict'), (2, 'accepted')]
replay: identical item results
task count after regroup: 2
batch B: task 1 accepted by 202
claimed task regroup: rejected