外卖系统选型中的重复通知与打印任务唯一性校验

原创

微订技术团队

修改于 2026-11-07 17:06:41

470

外卖系统选型时，技术评审应验证重复订单通知、打印任务登记和结果未知时的处理。本文使用SQLite事务及复合唯一约束，核对同请求重试、载荷冲突、跨商家归属和状态转换，给接单打印演示提供可复现的输入输出。

把采购问题限定为任务登记

构造场景中，商家接到已付款订单后登记打印任务。请求号相同且载荷一致时返回原任务，同请求号换订单或设备时拒绝。另一个请求号重复登记同订单、同设备、同用途时，也定位原任务。

示例用商家、请求号作为请求主键，商家、订单、设备和用途共同约束任务唯一。用途区分原始打印和经授权补打，示例只接受original。设备接收、实际出纸和回执恢复属于后续适配。

完整实现与复现

保存为print_task_revision.py，使用Python 3运行python print_task_revision.py，无第三方依赖。示例在内存SQLite中以单连接顺序执行。商家身份应来自认证上下文，正式服务还需实现授权、参数校验和持久化。

import sqlite3

db = sqlite3.connect(':memory:')
db.executescript('''
CREATE TABLE task (
    id INTEGER PRIMARY KEY,
    merchant TEXT NOT NULL,
    order_id TEXT NOT NULL,
    printer TEXT NOT NULL,
    purpose TEXT NOT NULL CHECK (purpose = 'original'),
    state TEXT NOT NULL DEFAULT 'pending' CHECK (state IN ('pending', 'unknown', 'confirmed')),
    UNIQUE (merchant, order_id, printer, purpose)
);
CREATE TABLE request (
    merchant TEXT NOT NULL,
    request_id TEXT NOT NULL,
    order_id TEXT NOT NULL,
    printer TEXT NOT NULL,
    purpose TEXT NOT NULL,
    task_id INTEGER NOT NULL REFERENCES task(id),
    PRIMARY KEY (merchant, request_id)
);
PRAGMA foreign_keys = ON;
''')


def register(merchant, request_id, order_id, printer, purpose='original'):
    if purpose != 'original':
        raise ValueError('purpose')
    payload = (order_id, printer, purpose)
    with db:
        previous = db.execute(
            'SELECT order_id, printer, purpose, task_id FROM request '
            'WHERE merchant = ? AND request_id = ?',
            (merchant, request_id)
        ).fetchone()
        if previous:
            if previous[:3] != payload:
                raise ValueError('request_conflict')
            return previous[3]
        db.execute(
            'INSERT INTO task(merchant, order_id, printer, purpose) '
            'VALUES (?, ?, ?, ?) '
            'ON CONFLICT(merchant, order_id, printer, purpose) DO NOTHING',
            (merchant, *payload)
        )
        task_id = db.execute(
            'SELECT id FROM task WHERE merchant = ? AND order_id = ? '
            'AND printer = ? AND purpose = ?',
            (merchant, *payload)
        ).fetchone()[0]
        db.execute('INSERT INTO request VALUES (?, ?, ?, ?, ?, ?)',
                   (merchant, request_id, *payload, task_id))
        return task_id


assert register('M', 'r1', 'O', 'P') == 1
assert register('M', 'r1', 'O', 'P') == 1
assert register('M', 'r2', 'O', 'P') == 1
for order, printer, reason in [('OTHER', 'P', 'request_conflict'),
                                ('O', 'OTHER', 'request_conflict')]:
    try:
        register('M', 'r1', order, printer)
        raise AssertionError('conflict accepted')
    except ValueError as error:
        assert str(error) == reason
assert db.execute('SELECT count(*) FROM task').fetchone()[0] == 1
assert register('N', 'r1', 'O', 'P') == 2
try:
    register('M', 'r3', 'O', 'P', 'reprint')
    raise AssertionError('unapproved reprint accepted')
except ValueError as error:
    assert str(error) == 'purpose'
assert db.execute('SELECT count(*) FROM request').fetchone()[0] == 3
print('9 assertions passed; tasks=2; request mappings=3')


def transition(task_id, before, after):
    allowed = {('pending', 'unknown'), ('pending', 'confirmed'),
               ('unknown', 'confirmed')}
    if (before, after) not in allowed:
        raise ValueError('transition')
    with db:
        return db.execute(
            'UPDATE task SET state = ? WHERE id = ? AND state = ?',
            (after, task_id, before)
        ).rowcount


assert transition(1, 'pending', 'unknown') == 1
assert db.execute('SELECT state FROM task WHERE id = 1').fetchone()[0] == 'unknown'
assert transition(1, 'unknown', 'confirmed') == 1
assert transition(1, 'pending', 'unknown') == 0
assert db.execute('SELECT state FROM task WHERE id = 1').fetchone()[0] == 'confirmed'
assert transition(999, 'pending', 'confirmed') == 0
try:
    transition(1, 'confirmed', 'pending')
    raise AssertionError('terminal rollback accepted')
except ValueError as error:
    assert str(error) == 'transition'
print('16 assertions passed in total; task 1 confirmed; stale update=0')