From 12a5c998363bea8404858d12f333d6f1fd1d6188 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=83=E7=B5=B2=E7=90=AA?= Date: Sat, 12 Sep 2026 17:12:56 +0800 Subject: [PATCH] Make sales order inventory updates atomic --- backend/orders.py | 287 +++++++++++++++++++++--- frontend/page_sales.py | 57 +++-- tests/test_order_inventory_atomicity.py | 139 ++++++++++++ 3 files changed, 434 insertions(+), 49 deletions(-) create mode 100644 tests/test_order_inventory_atomicity.py diff --git a/backend/orders.py b/backend/orders.py index 9f39c3f..610aaf2 100644 --- a/backend/orders.py +++ b/backend/orders.py @@ -4,10 +4,247 @@ """ from datetime import datetime -from .database import run_query +from uuid import uuid4 + +from .database import run_query, transaction from .auth import check_permission +ORDER_STATUSES = frozenset({"處理中", "已出貨", "已取消"}) +_STOCK_RESERVED_STATUSES = frozenset({"處理中", "已出貨"}) + + +class OrderValidationError(ValueError): + """The requested order operation is not valid.""" + + +class InsufficientStockError(OrderValidationError): + """The order cannot reserve the requested stock.""" + + +def _validate_positive_quantity(quantity: int) -> int: + if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0: + raise OrderValidationError("訂單數量必須是正整數。") + return quantity + + +def _validate_status(status: str) -> str: + normalized = str(status or "").strip() + if normalized not in ORDER_STATUSES: + raise OrderValidationError(f"不支援的訂單狀態：{normalized or '空白'}。") + return normalized + + +def _write_stock_move( + conn, + *, + product_id: str, + warehouse_id: str | None, + quantity_change: int, + move_type: str, + order_id: str, + move_date: str, +) -> None: + conn.execute( + """ + INSERT INTO stock_moves ( + product_id, warehouse_id, qty, move_type, ref_no, move_date, note + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + product_id, + warehouse_id, + quantity_change, + move_type, + order_id, + move_date, + "銷售訂單庫存異動", + ), + ) + + +def create_sales_order( + *, + order_id: str, + customer_id: str | None, + product_id: str, + quantity: int, + status: str = "處理中", + order_date: str | None = None, +) -> dict: + """Create an order and reserve stock in one atomic transaction.""" + order_id = str(order_id or "").strip() + product_id = str(product_id or "").strip() + customer_id = str(customer_id or "").strip() or None + if not order_id: + raise OrderValidationError("訂單編號不可空白。") + if not product_id: + raise OrderValidationError("產品編號不可空白。") + quantity = _validate_positive_quantity(quantity) + status = _validate_status(status) + order_date = str(order_date or "").strip() or datetime.now().strftime( + "%Y-%m-%d %H:%M:%S" + ) + + with transaction(immediate=True) as conn: + product = conn.execute( + "SELECT name, stock, price, warehouse_id FROM inventory WHERE product_id = ?", + (product_id,), + ).fetchone() + if product is None: + raise OrderValidationError(f"找不到產品編號 {product_id}。") + name, current_stock, price, warehouse_id = product + + if customer_id is not None and conn.execute( + "SELECT 1 FROM customers WHERE customer_id = ?", (customer_id,) + ).fetchone() is None: + raise OrderValidationError(f"找不到客戶編號 {customer_id}。") + + remaining_stock = current_stock + if status in _STOCK_RESERVED_STATUSES: + updated = conn.execute( + """ + UPDATE inventory + SET stock = stock - ? + WHERE product_id = ? AND stock >= ? + """, + (quantity, product_id, quantity), + ) + if updated.rowcount != 1: + raise InsufficientStockError( + f"庫存不足！產品 {name} 目前只有 {current_stock} 件，" + f"無法售出 {quantity} 件。" + ) + remaining_stock = current_stock - quantity + + conn.execute( + """ + INSERT INTO orders ( + order_id, customer_id, product_id, quantity, status, + order_date, total_amount + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + order_id, + customer_id, + product_id, + quantity, + status, + order_date, + quantity * (price or 0), + ), + ) + if status in _STOCK_RESERVED_STATUSES: + _write_stock_move( + conn, + product_id=product_id, + warehouse_id=warehouse_id, + quantity_change=-quantity, + move_type="銷售預留", + order_id=order_id, + move_date=order_date, + ) + + return { + "order_id": order_id, + "product_id": product_id, + "product_name": name, + "quantity": quantity, + "status": status, + "remaining_stock": remaining_stock, + "order_date": order_date, + } + + +def transition_order_status(order_id: str, new_status: str) -> dict: + """Change status and apply exactly one matching stock effect atomically.""" + order_id = str(order_id or "").strip() + if not order_id: + raise OrderValidationError("訂單編號不可空白。") + new_status = _validate_status(new_status) + changed_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + with transaction(immediate=True) as conn: + row = conn.execute( + """ + SELECT o.product_id, o.quantity, o.status, i.name, i.stock, + i.warehouse_id + FROM orders o + JOIN inventory i ON i.product_id = o.product_id + WHERE o.order_id = ? + """, + (order_id,), + ).fetchone() + if row is None: + raise OrderValidationError(f"找不到訂單 {order_id}。") + + product_id, quantity, old_status, name, current_stock, warehouse_id = row + if old_status == new_status: + return { + "order_id": order_id, + "old_status": old_status, + "new_status": new_status, + "remaining_stock": current_stock, + "changed": False, + } + + old_reserved = old_status in _STOCK_RESERVED_STATUSES + new_reserved = new_status in _STOCK_RESERVED_STATUSES + remaining_stock = current_stock + if old_reserved and not new_reserved: + conn.execute( + "UPDATE inventory SET stock = stock + ? WHERE product_id = ?", + (quantity, product_id), + ) + remaining_stock = current_stock + quantity + _write_stock_move( + conn, + product_id=product_id, + warehouse_id=warehouse_id, + quantity_change=quantity, + move_type="取消回補", + order_id=order_id, + move_date=changed_at, + ) + elif not old_reserved and new_reserved: + updated = conn.execute( + """ + UPDATE inventory + SET stock = stock - ? + WHERE product_id = ? AND stock >= ? + """, + (quantity, product_id, quantity), + ) + if updated.rowcount != 1: + raise InsufficientStockError( + f"庫存不足！產品 {name} 目前只有 {current_stock} 件，" + f"無法重新啟用數量 {quantity} 的訂單。" + ) + remaining_stock = current_stock - quantity + _write_stock_move( + conn, + product_id=product_id, + warehouse_id=warehouse_id, + quantity_change=-quantity, + move_type="重新預留", + order_id=order_id, + move_date=changed_at, + ) + + conn.execute( + "UPDATE orders SET status = ? WHERE order_id = ?", + (new_status, order_id), + ) + + return { + "order_id": order_id, + "old_status": old_status, + "new_status": new_status, + "remaining_stock": remaining_stock, + "changed": True, + } + + def get_recent_orders() -> str: """查詢近期所有訂單（包含訂單建立日期）""" if not check_permission(["sales", "warehouse"]): @@ -40,34 +277,26 @@ def create_order(product_id: str, quantity: int) -> str: if quantity <= 0: return "建立失敗：訂單數量必須 > 0。" - # 先檢查庫存是否足夠 - res = run_query("SELECT stock, name FROM inventory WHERE product_id=?", (product_id,)) - if not res: - return f"建立失敗：找不到產品編號 {product_id}。" - - current_stock, name = res[0] - if current_stock < quantity: - return f"建立失敗：庫存不足！產品 {name} 目前只有 {current_stock} 件，無法售出 {quantity} 件。" - - # 產生訂單號碼與時間 now_dt = datetime.now() - order_id = f"ORD-{now_dt.strftime('%Y%m%d-%H%M%S')}" + order_id = f"ORD-{now_dt.strftime('%Y%m%d-%H%M%S-%f')}-{uuid4().hex[:6]}" order_date = now_dt.strftime('%Y-%m-%d %H:%M:%S') - res_price = run_query("SELECT price FROM inventory WHERE product_id=?", (product_id,)) - total_amount = (quantity * res_price[0][0]) if res_price else 0 try: - run_query( - "INSERT INTO orders (order_id, customer_id, product_id, quantity, status, order_date, total_amount) VALUES (?, ?, ?, ?, ?, ?, ?)", - (order_id, None, product_id, quantity, "處理中", order_date, total_amount), - fetch=False, + result = create_sales_order( + order_id=order_id, + customer_id=None, + product_id=product_id, + quantity=quantity, + status="處理中", + order_date=order_date, ) - # 扣除庫存 - run_query("UPDATE inventory SET stock=? WHERE product_id=?", (current_stock - quantity, product_id), fetch=False) return ( f"✅ 成功建立新訂單 (單號: {order_id}，時間: {order_date})：" - f"售出 {quantity} 件 {name} ({product_id})，庫存已自動扣除，目前剩餘 {current_stock - quantity} 件。" + f"售出 {quantity} 件 {result['product_name']} ({product_id})，" + f"庫存已自動扣除，目前剩餘 {result['remaining_stock']} 件。" ) + except OrderValidationError as e: + return f"建立失敗：{e}" except Exception as e: return f"建立訂單時發生資料庫錯誤：{e}" @@ -83,12 +312,12 @@ def cancel_order(product_id: str, quantity: int, customer_id: str = "") -> str: if customer_id: rows = run_query( - "SELECT order_id FROM orders WHERE (customer_id = ? OR customer_id IS NULL) AND product_id = ? AND quantity = ? ORDER BY order_date DESC LIMIT 1", + "SELECT order_id FROM orders WHERE (customer_id = ? OR customer_id IS NULL) AND product_id = ? AND quantity = ? AND status != '已取消' ORDER BY order_date DESC LIMIT 1", (customer_id, product_id, quantity), ) else: rows = run_query( - "SELECT order_id FROM orders WHERE product_id = ? AND quantity = ? ORDER BY order_date DESC LIMIT 1", + "SELECT order_id FROM orders WHERE product_id = ? AND quantity = ? AND status != '已取消' ORDER BY order_date DESC LIMIT 1", (product_id, quantity), ) @@ -96,10 +325,14 @@ def cancel_order(product_id: str, quantity: int, customer_id: str = "") -> str: return f"找不到產品 {product_id} 數量 {quantity} 的待取消訂單。" order_id = rows[0][0] - run_query("DELETE FROM orders WHERE order_id = ?", (order_id,), fetch=False) - from backend.inventory import update_inventory - update_inventory(product_id=product_id, quantity_change=quantity) - return f"✅ 已取消訂單 {order_id}，並將產品 {product_id} 庫存回補 {quantity} 件。" + try: + result = transition_order_status(order_id, "已取消") + except OrderValidationError as exc: + return f"取消訂單失敗：{exc}" + return ( + f"✅ 已取消訂單 {order_id}，並將產品 {product_id} 庫存回補 " + f"{quantity} 件，目前庫存 {result['remaining_stock']} 件。" + ) def get_receivables() -> str: diff --git a/frontend/page_sales.py b/frontend/page_sales.py index 08c5b44..2820557 100644 --- a/frontend/page_sales.py +++ b/frontend/page_sales.py @@ -9,6 +9,12 @@ import plotly.express as px from datetime import datetime from backend import DB_FILE, run_query +from backend.orders import ( + InsufficientStockError, + OrderValidationError, + create_sales_order, + transition_order_status, +) def render(sub_menu: str, api_key: str): @@ -67,26 +73,24 @@ def render(sub_menu: str, api_key: str): o_qty = st.number_input("數量", min_value=1) o_status = st.selectbox("狀態", ["處理中", "已出貨", "已取消"]) if st.form_submit_button("送出") and o_id and p_id: - res = run_query("SELECT stock, name, price FROM inventory WHERE product_id=?", (p_id,)) - if not res: - st.error("找不到產品") - else: - stock, pname, up = res[0] - total = o_qty * up - if o_status != "已取消" and stock < o_qty: - st.error(f"庫存不足，目前 {stock} 件") - else: - try: - run_query( - "INSERT INTO orders (order_id, customer_id, product_id, quantity, status, order_date, total_amount) VALUES (?,?,?,?,?,?,?)", - (o_id, cust, p_id, o_qty, o_status, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), total), - fetch=False, - ) - if o_status != "已取消": - run_query("UPDATE inventory SET stock=? WHERE product_id=?", (stock - o_qty, p_id), fetch=False) - st.success(f"訂單 {o_id} 已建立，金額 {total:,.0f} 元") - except sqlite3.IntegrityError: - st.error("訂單編號已存在") + try: + result = create_sales_order( + order_id=o_id, + customer_id=cust, + product_id=p_id, + quantity=int(o_qty), + status=o_status, + ) + st.success( + f"訂單 {o_id} 已建立，目前庫存 " + f"{result['remaining_stock']} 件" + ) + except InsufficientStockError as exc: + st.error(str(exc)) + except OrderValidationError as exc: + st.error(str(exc)) + except sqlite3.IntegrityError: + st.error("訂單編號已存在") with st.expander("🔄 更新訂單狀態（消除沙漏 ⏳／逾期 🚨）"): st.caption("警示說明：⏳ 處理中（未滿 3 天）｜🚨 逾期（處理中超過 3 天）｜✅ 已出貨／已取消。將訂單改為「已出貨」或「已取消」後，警示會顯示 ✅。") ord_list = run_query("SELECT order_id, status FROM orders ORDER BY order_date DESC LIMIT 100") @@ -96,8 +100,17 @@ def render(sub_menu: str, api_key: str): sel_ord = st.selectbox("選擇訂單", list(ord_opts.keys()), format_func=lambda x: ord_opts.get(x, x)) new_status = st.selectbox("新狀態", ["已出貨", "已取消", "處理中"]) if st.form_submit_button("更新狀態") and sel_ord: - run_query("UPDATE orders SET status=? WHERE order_id=?", (new_status, sel_ord), fetch=False) - st.success(f"訂單 {sel_ord} 已更新為「{new_status}」。重新整理後警示將顯示 ✅。") + try: + result = transition_order_status(sel_ord, new_status) + if result["changed"]: + st.success( + f"訂單 {sel_ord} 已更新為「{new_status}」，" + f"目前庫存 {result['remaining_stock']} 件。" + ) + else: + st.info(f"訂單 {sel_ord} 已經是「{new_status}」。") + except (InsufficientStockError, OrderValidationError) as exc: + st.error(str(exc)) else: st.info("尚無訂單") # ── 銷售單查詢與篩選面板 (可摺疊) ── diff --git a/tests/test_order_inventory_atomicity.py b/tests/test_order_inventory_atomicity.py new file mode 100644 index 0000000..798214b --- /dev/null +++ b/tests/test_order_inventory_atomicity.py @@ -0,0 +1,139 @@ +"""Sales-order and inventory effects must commit atomically.""" + +import sqlite3 +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from backend import database +from backend.orders import ( + InsufficientStockError, + create_sales_order, + transition_order_status, +) + + +@pytest.fixture(autouse=True) +def initialized_db(): + database.init_db() + database.run_query("UPDATE inventory SET stock = 10 WHERE product_id = 'P001'", fetch=False) + database.run_query("DELETE FROM stock_moves WHERE product_id = 'P001'", fetch=False) + database.run_query("DELETE FROM orders WHERE order_id LIKE 'ATOMIC-%'", fetch=False) + + +def _stock() -> int: + return database.run_query( + "SELECT stock FROM inventory WHERE product_id = 'P001'" + )[0][0] + + +def test_create_order_commits_order_stock_and_movement_together(): + result = create_sales_order( + order_id="ATOMIC-CREATE", + customer_id="C001", + product_id="P001", + quantity=3, + ) + + assert result["remaining_stock"] == 7 + assert _stock() == 7 + assert database.run_query( + "SELECT status FROM orders WHERE order_id = 'ATOMIC-CREATE'" + ) == [("處理中",)] + assert database.run_query( + "SELECT qty, move_type, ref_no FROM stock_moves WHERE ref_no = 'ATOMIC-CREATE'" + ) == [(-3, "銷售預留", "ATOMIC-CREATE")] + + +def test_duplicate_order_id_rolls_back_stock_change(): + create_sales_order( + order_id="ATOMIC-DUPLICATE", + customer_id="C001", + product_id="P001", + quantity=2, + ) + stock_before_retry = _stock() + + with pytest.raises(sqlite3.IntegrityError): + create_sales_order( + order_id="ATOMIC-DUPLICATE", + customer_id="C001", + product_id="P001", + quantity=2, + ) + + assert _stock() == stock_before_retry + assert database.run_query( + "SELECT COUNT(*) FROM stock_moves WHERE ref_no = 'ATOMIC-DUPLICATE'" + )[0][0] == 1 + + +def test_concurrent_orders_cannot_oversell(): + def submit(order_id: str): + try: + create_sales_order( + order_id=order_id, + customer_id="C001", + product_id="P001", + quantity=7, + ) + return "created" + except InsufficientStockError: + return "insufficient" + + with ThreadPoolExecutor(max_workers=2) as pool: + outcomes = list(pool.map(submit, ("ATOMIC-RACE-A", "ATOMIC-RACE-B"))) + + assert sorted(outcomes) == ["created", "insufficient"] + assert _stock() == 3 + assert database.run_query( + "SELECT COUNT(*) FROM orders WHERE order_id IN ('ATOMIC-RACE-A', 'ATOMIC-RACE-B')" + )[0][0] == 1 + + +def test_cancel_is_idempotent_and_reactivation_reserves_again(): + create_sales_order( + order_id="ATOMIC-STATUS", + customer_id="C001", + product_id="P001", + quantity=4, + ) + assert _stock() == 6 + + cancelled = transition_order_status("ATOMIC-STATUS", "已取消") + assert cancelled["remaining_stock"] == 10 + assert _stock() == 10 + + repeated = transition_order_status("ATOMIC-STATUS", "已取消") + assert repeated["changed"] is False + assert _stock() == 10 + + reactivated = transition_order_status("ATOMIC-STATUS", "處理中") + assert reactivated["remaining_stock"] == 6 + assert _stock() == 6 + assert database.run_query( + "SELECT qty, move_type FROM stock_moves WHERE ref_no = 'ATOMIC-STATUS' ORDER BY move_id" + ) == [ + (-4, "銷售預留"), + (4, "取消回補"), + (-4, "重新預留"), + ] + + +def test_failed_reactivation_keeps_cancelled_status_and_stock(): + create_sales_order( + order_id="ATOMIC-REACTIVATE", + customer_id="C001", + product_id="P001", + quantity=8, + ) + transition_order_status("ATOMIC-REACTIVATE", "已取消") + database.run_query("UPDATE inventory SET stock = 2 WHERE product_id = 'P001'", fetch=False) + + with pytest.raises(InsufficientStockError): + transition_order_status("ATOMIC-REACTIVATE", "已出貨") + + assert _stock() == 2 + assert database.run_query( + "SELECT status FROM orders WHERE order_id = 'ATOMIC-REACTIVATE'" + ) == [("已取消",)]