printedparts stage 5: the ledger - restock/adjust with badge attribution
Some checks failed
CI / backend (push) Failing after 1m42s
CI / naming (push) Successful in 1s
CI / frontend (push) Successful in 8s
CI / migrations-mysql (push) Failing after 8s

Badge resolver copied from the USB contract (SSO digits, 0<digits>BZ
PayNo wrap) with names from the employees directory and the
unknown-badge policy setting; deliberately copied rather than
cross-imported so the contract test stays green. Restock and adjust
write the ledger row and move the cached quantity in one commit -
the single-commit invariant every write path must use. Adjust
requires a reason and refuses to drive stock below zero. Detail page
gains Restock/Adjust modals. Seven tests cover minting, the
cache==ledger invariant, badge shapes, policy toggle, and auth.
This commit is contained in:
cproudlock
2026-07-17 07:41:18 -04:00
parent cb367a38f9
commit 6dfc8906c4
6 changed files with 349 additions and 0 deletions

View File

@@ -210,3 +210,80 @@ def delete_item_image(item_id: int):
item.imageurl = None
db.session.commit()
return success_response(item.to_dict(), message='Item image removed')
# --- the ledger: restock and adjust (stage 6 gates with printedparts.restock)
from ..models import PrintedItemTransaction
from ..services.badges import BadgeError, resolve_badge
def _ledger_write(item, transactiontype, quantitychange, sso, name, reason=None):
"""Append a ledger row and move the cached quantity in ONE commit.
The single-commit invariant is what keeps quantityonhand equal to the
ledger sum; every write path must go through here.
"""
item.quantityonhand += quantitychange
db.session.add(PrintedItemTransaction(
printeditemid=item.printeditemid,
transactiontype=transactiontype,
quantitychange=quantitychange,
employeesso=sso,
employeename=name,
reason=reason,
))
db.session.commit()
@printedparts_bp.route('/items/<int:item_id>/restock', methods=['POST'])
@jwt_required()
def restock_item(item_id: int):
"""Add freshly printed stock. Body: {quantity, badge}."""
item = db.session.get(PrintedItem, item_id)
if not item or not item.isactive:
return error_response(ErrorCodes.NOT_FOUND,
f'Printed item {item_id} not found', http_code=404)
data = request.get_json() or {}
quantity = data.get('quantity')
if not isinstance(quantity, int) or quantity < 1:
return error_response(ErrorCodes.VALIDATION_ERROR,
'quantity must be a positive integer')
try:
sso, name = resolve_badge(data.get('badge'))
except BadgeError as badge_error:
return error_response(ErrorCodes.VALIDATION_ERROR, str(badge_error),
http_code=422)
_ledger_write(item, 'restock', quantity, sso, name)
return success_response(item.to_dict(), message='Stock added')
@printedparts_bp.route('/items/<int:item_id>/adjust', methods=['POST'])
@jwt_required()
def adjust_item(item_id: int):
"""Correct the count (damage, recount). Body: {quantitychange, reason, badge}."""
item = db.session.get(PrintedItem, item_id)
if not item or not item.isactive:
return error_response(ErrorCodes.NOT_FOUND,
f'Printed item {item_id} not found', http_code=404)
data = request.get_json() or {}
quantitychange = data.get('quantitychange')
if not isinstance(quantitychange, int) or quantitychange == 0:
return error_response(ErrorCodes.VALIDATION_ERROR,
'quantitychange must be a non-zero integer')
reason = (data.get('reason') or '').strip()
if not reason:
return error_response(ErrorCodes.VALIDATION_ERROR,
'reason is required for an adjustment')
if item.quantityonhand + quantitychange < 0:
return error_response(
ErrorCodes.VALIDATION_ERROR,
f'Adjustment would drive stock below zero '
f'(on hand: {item.quantityonhand})')
try:
sso, name = resolve_badge(data.get('badge'))
except BadgeError as badge_error:
return error_response(ErrorCodes.VALIDATION_ERROR, str(badge_error),
http_code=422)
_ledger_write(item, 'adjust', quantitychange, sso, name, reason=reason)
return success_response(item.to_dict(), message='Stock adjusted')