print_queue.py 103 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452
  1. """API routes for print queue management."""
  2. import json
  3. import logging
  4. import zipfile
  5. from datetime import datetime, timezone
  6. from pathlib import Path
  7. import defusedxml.ElementTree as ET
  8. from fastapi import APIRouter, Depends, HTTPException, Query
  9. from sqlalchemy import and_, func, inspect, or_, select, update
  10. from sqlalchemy.ext.asyncio import AsyncSession
  11. from sqlalchemy.orm import selectinload
  12. from backend.app.api.routes.library_variants import normalize_model_name, resolve_variant_model
  13. from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_ownership_permission
  14. from backend.app.core.config import settings
  15. from backend.app.core.database import get_db
  16. from backend.app.core.permissions import Permission
  17. from backend.app.models.archive import PrintArchive
  18. from backend.app.models.library import LibraryFile
  19. from backend.app.models.print_batch import PrintBatch, PrintBatchPlate
  20. from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
  21. from backend.app.models.printer import Printer
  22. from backend.app.models.project import Project
  23. from backend.app.models.user import User
  24. from backend.app.schemas.print_queue import (
  25. PrintBatchCreate,
  26. PrintBatchDispatchRequest,
  27. PrintBatchPlateProgress,
  28. PrintBatchPlateTarget,
  29. PrintBatchResponse,
  30. PrintBatchUngroupResponse,
  31. PrintBatchUpdate,
  32. PrintQueueBulkUpdate,
  33. PrintQueueBulkUpdateResponse,
  34. PrintQueueItemCreate,
  35. PrintQueueItemResponse,
  36. PrintQueueItemUpdate,
  37. PrintQueueReorder,
  38. QueueVariantCreate,
  39. QueueVariantSummary,
  40. )
  41. from backend.app.services.filament_deficit import compute_deficit_for_queue_item
  42. from backend.app.services.filament_requirements import overrides_for_plate
  43. from backend.app.services.finance_budget import release_budget_reservation, validate_print_budget
  44. from backend.app.services.notification_service import notification_service
  45. from backend.app.services.print_batch import (
  46. BatchDispatchError,
  47. dispatch_remaining,
  48. load_progress,
  49. refresh_batch_status,
  50. refresh_batch_status_for_item,
  51. )
  52. from backend.app.services.print_cost_estimate import estimate_queue_source_cost
  53. from backend.app.utils.printer_models import (
  54. is_gcode_compatible,
  55. )
  56. from backend.app.utils.threemf_tools import (
  57. extract_plate_metadata_from_3mf,
  58. extract_print_time_from_3mf,
  59. )
  60. logger = logging.getLogger(__name__)
  61. router = APIRouter(prefix="/queue", tags=["queue"])
  62. def _variant_summaries(item: PrintQueueItem) -> list[QueueVariantSummary]:
  63. """Cross-model candidates for display (#671), or [] if they weren't loaded.
  64. Every route that builds a queue response eager-loads ``variants``. Reading
  65. the attribute unguarded would still be a landmine for the next one that
  66. doesn't: a lazy load on an async session raises rather than degrading, so a
  67. forgotten ``selectinload`` would turn a card render into a 500.
  68. """
  69. if "variants" in inspect(item).unloaded:
  70. return []
  71. return [
  72. QueueVariantSummary(
  73. library_file_id=v.library_file_id,
  74. filename=v.library_file.filename if v.library_file else "",
  75. target_model=v.target_model,
  76. position=v.position,
  77. )
  78. for v in item.variants
  79. ]
  80. def _extract_filament_types_from_3mf(file_path: Path, plate_id: int | None = None) -> list[str]:
  81. """Extract unique filament types from a 3MF file.
  82. Args:
  83. file_path: Path to the 3MF file
  84. plate_id: Optional plate index to filter for (for multi-plate files)
  85. Returns:
  86. List of unique filament types (e.g., ["PLA", "PETG"])
  87. """
  88. types: set[str] = set()
  89. try:
  90. with zipfile.ZipFile(file_path, "r") as zf:
  91. if "Metadata/slice_info.config" not in zf.namelist():
  92. return []
  93. content = zf.read("Metadata/slice_info.config").decode()
  94. root = ET.fromstring(content)
  95. if plate_id is not None:
  96. # Find the plate element with matching index
  97. for plate_elem in root.findall(".//plate"):
  98. plate_index = None
  99. for meta in plate_elem.findall("metadata"):
  100. if meta.get("key") == "index":
  101. try:
  102. plate_index = int(meta.get("value", "0"))
  103. except ValueError:
  104. pass # Skip plate with unparseable index
  105. break
  106. if plate_index == plate_id:
  107. for filament_elem in plate_elem.findall("filament"):
  108. filament_type = filament_elem.get("type", "")
  109. used_g = filament_elem.get("used_g", "0")
  110. try:
  111. used_grams = float(used_g)
  112. except (ValueError, TypeError):
  113. used_grams = 0
  114. if used_grams > 0 and filament_type:
  115. types.add(filament_type)
  116. break
  117. else:
  118. # No plate_id specified - extract all filaments with used_g > 0
  119. for filament_elem in root.findall(".//filament"):
  120. filament_type = filament_elem.get("type", "")
  121. used_g = filament_elem.get("used_g", "0")
  122. try:
  123. used_grams = float(used_g)
  124. except (ValueError, TypeError):
  125. used_grams = 0
  126. if used_grams > 0 and filament_type:
  127. types.add(filament_type)
  128. except Exception as e:
  129. logger.warning("Failed to extract filament types from %s: %s", file_path, e)
  130. return sorted(types)
  131. # Local alias kept so existing call sites stay compact; the implementation lives
  132. # in utils/threemf_tools.py so the notification path (main.py) can reuse it
  133. # without importing from a routes module (#1785).
  134. _extract_print_time_from_3mf = extract_print_time_from_3mf
  135. def _assert_can_queue_archive(archive: PrintArchive, current_user: User | None) -> None:
  136. """Gate turning *archive* into a print. Raises rather than returning a verdict.
  137. Shared by every route that creates queue items from an archive, so a new
  138. one can't quietly become a weaker door to the same action than
  139. ``POST /queue/`` is.
  140. Two separate checks:
  141. * IDOR fix (maziggy/bambuddy-security #2): without this, a caller with
  142. QUEUE_CREATE could queue any user's archive even without ARCHIVES_READ on
  143. it — Landon's PoC enumerated this on admin's archives as operator1. Gate
  144. on ARCHIVES_READ_ALL OR ownership. 404 (not 403) so we don't leak "this
  145. id exists but you can't queue it" for enumeration.
  146. * Reprint perm gate (#1625): the legacy ``/archives/{id}/reprint`` endpoint
  147. required ARCHIVES_REPRINT_OWN/ALL, and every route that replaces it must
  148. keep that gate or an operator with QUEUE_CREATE could reprint via a
  149. direct API call even when explicitly denied reprint perm. Mirrors the
  150. frontend ``canModify('archives', 'reprint', ...)`` helper: REPRINT_ALL
  151. allows any archive, REPRINT_OWN allows own only, ownerless archives
  152. require REPRINT_ALL (fail-closed).
  153. """
  154. if current_user is None:
  155. return
  156. if not current_user.has_permission(Permission.ARCHIVES_READ_ALL.value) and archive.created_by_id != current_user.id:
  157. raise HTTPException(404, "Archive not found")
  158. owns_archive = archive.created_by_id is not None and archive.created_by_id == current_user.id
  159. has_reprint = current_user.has_permission(Permission.ARCHIVES_REPRINT_ALL.value) or (
  160. owns_archive and current_user.has_permission(Permission.ARCHIVES_REPRINT_OWN.value)
  161. )
  162. if not has_reprint:
  163. raise HTTPException(
  164. status_code=403,
  165. detail="Permission archives:reprint_own or archives:reprint_all required",
  166. )
  167. def _assert_can_queue_library_file(library_file: LibraryFile, current_user: User | None) -> None:
  168. """Gate turning *library_file* into a print — LIBRARY_READ_ALL or ownership."""
  169. if current_user is None:
  170. return
  171. if (
  172. not current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
  173. and library_file.created_by_id != current_user.id
  174. ):
  175. raise HTTPException(404, "Library file not found")
  176. async def _is_orders_last_source(db: AsyncSession, item: PrintQueueItem) -> bool:
  177. """True when deleting *item* would leave an order owing runs it can't queue.
  178. Dispatch produces the runs an order still owes by cloning an existing
  179. queue item for the same plate — that row is the only record of the printer
  180. target, AMS mapping and print options the user chose. Delete the last one
  181. while the plate still has a target and the order is stranded: it goes on
  182. reporting work outstanding with no way left to produce it (#2960).
  183. """
  184. if item.batch_id is None:
  185. return False
  186. # A cancelled order can never dispatch again, so nothing about it can be
  187. # stranded and its leftover rows must stay deletable — tidying up after a
  188. # cancel is the most likely reason anyone deletes them.
  189. status = (await db.execute(select(PrintBatch.status).where(PrintBatch.id == item.batch_id))).scalar_one_or_none()
  190. if status == "cancelled":
  191. return False
  192. plate_scope = (
  193. PrintBatchPlate.plate_id == item.plate_id if item.plate_id is not None else PrintBatchPlate.plate_id.is_(None)
  194. )
  195. # first(), not scalar_one_or_none(): a UNIQUE(batch_id, plate_id) does not
  196. # constrain NULL plate_ids on either dialect, and a duplicate whole-file
  197. # row must not turn a delete into a 500.
  198. target = (
  199. (
  200. await db.execute(
  201. select(PrintBatchPlate.quantity_target)
  202. .where(PrintBatchPlate.batch_id == item.batch_id)
  203. .where(plate_scope)
  204. .order_by(PrintBatchPlate.quantity_target.desc())
  205. .limit(1)
  206. )
  207. )
  208. .scalars()
  209. .first()
  210. )
  211. # No target row: a grouping, or a plate assigned into the order by hand.
  212. # A target of 0 is legal and means the plate is not wanted. Neither owes
  213. # anything, so neither can be stranded.
  214. if not target:
  215. return False
  216. item_scope = (
  217. PrintQueueItem.plate_id == item.plate_id if item.plate_id is not None else PrintQueueItem.plate_id.is_(None)
  218. )
  219. survivor = (
  220. await db.execute(
  221. select(PrintQueueItem.id)
  222. .where(PrintQueueItem.batch_id == item.batch_id)
  223. .where(item_scope)
  224. .where(PrintQueueItem.id != item.id)
  225. .limit(1)
  226. )
  227. ).scalar_one_or_none()
  228. return survivor is None
  229. async def _assert_can_dispatch_batch_sources(db: AsyncSession, batch_id: int, current_user: User | None) -> None:
  230. """Apply the ``POST /queue/`` source-file gates to everything a dispatch would print.
  231. Dispatching clones existing queue items, so without this it would be a
  232. weaker door to the same outcome: a caller holding QUEUE_CREATE and
  233. QUEUE_UPDATE_ALL but explicitly denied ``archives:reprint_*`` could start
  234. prints through an order that ``POST /queue/`` would have refused them.
  235. Every distinct source among the batch's items is checked, including the
  236. library files behind cross-model variants — those get cloned too, and any
  237. one of them may be the file that actually runs.
  238. """
  239. archive_ids = set(
  240. (
  241. await db.execute(
  242. select(PrintQueueItem.archive_id)
  243. .where(PrintQueueItem.batch_id == batch_id)
  244. .where(PrintQueueItem.archive_id.is_not(None))
  245. .distinct()
  246. )
  247. )
  248. .scalars()
  249. .all()
  250. )
  251. library_file_ids = set(
  252. (
  253. await db.execute(
  254. select(PrintQueueItem.library_file_id)
  255. .where(PrintQueueItem.batch_id == batch_id)
  256. .where(PrintQueueItem.library_file_id.is_not(None))
  257. .distinct()
  258. )
  259. )
  260. .scalars()
  261. .all()
  262. )
  263. library_file_ids |= set(
  264. (
  265. await db.execute(
  266. select(PrintQueueVariant.library_file_id)
  267. .join(PrintQueueItem, PrintQueueVariant.queue_item_id == PrintQueueItem.id)
  268. .where(PrintQueueItem.batch_id == batch_id)
  269. .distinct()
  270. )
  271. )
  272. .scalars()
  273. .all()
  274. )
  275. for archive_id in archive_ids:
  276. archive = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none()
  277. # A deleted source can't be printed; dispatch will fail on it anyway.
  278. if archive is not None:
  279. _assert_can_queue_archive(archive, current_user)
  280. for library_file_id in library_file_ids:
  281. library_file = (
  282. await db.execute(LibraryFile.active().where(LibraryFile.id == library_file_id))
  283. ).scalar_one_or_none()
  284. if library_file is not None:
  285. _assert_can_queue_library_file(library_file, current_user)
  286. async def _resolve_source_path(db: AsyncSession, item: PrintQueueItem) -> Path | None:
  287. """Resolve an existing queue item's source 3MF on disk, or None."""
  288. if item.archive_id:
  289. result = await db.execute(select(PrintArchive).where(PrintArchive.id == item.archive_id))
  290. archive = result.scalar_one_or_none()
  291. if archive:
  292. return settings.base_dir / archive.file_path
  293. elif item.library_file_id:
  294. result = await db.execute(LibraryFile.active().where(LibraryFile.id == item.library_file_id))
  295. library_file = result.scalar_one_or_none()
  296. if library_file:
  297. lib_path = Path(library_file.file_path)
  298. return lib_path if lib_path.is_absolute() else settings.base_dir / library_file.file_path
  299. return None
  300. async def _trusted_item_estimated_cost(
  301. db: AsyncSession,
  302. item: PrintQueueItem,
  303. *,
  304. printer_id: int | None,
  305. plate_id: int | None,
  306. ams_mapping: list[int] | str | None,
  307. ) -> float | None:
  308. """Recompute an existing queue item's cost from its persisted source."""
  309. if item.archive_id:
  310. archive = await db.scalar(select(PrintArchive).where(PrintArchive.id == item.archive_id))
  311. return await estimate_queue_source_cost(
  312. db,
  313. archive=archive,
  314. plate_id=plate_id,
  315. ams_mapping=ams_mapping,
  316. printer_id=printer_id,
  317. )
  318. if item.library_file_id:
  319. library_file = await db.scalar(LibraryFile.active().where(LibraryFile.id == item.library_file_id))
  320. return await estimate_queue_source_cost(
  321. db,
  322. library_file=library_file,
  323. plate_id=plate_id,
  324. ams_mapping=ams_mapping,
  325. printer_id=printer_id,
  326. )
  327. result = await db.execute(
  328. select(PrintQueueVariant, LibraryFile)
  329. .join(LibraryFile, LibraryFile.id == PrintQueueVariant.library_file_id)
  330. .where(PrintQueueVariant.queue_item_id == item.id)
  331. )
  332. estimates = [
  333. await estimate_queue_source_cost(
  334. db,
  335. library_file=library_file,
  336. plate_id=variant.plate_id,
  337. ams_mapping=variant.ams_mapping,
  338. printer_id=printer_id,
  339. )
  340. for variant, library_file in result.all()
  341. ]
  342. if not estimates or any(cost is None for cost in estimates):
  343. return None
  344. return max(cost for cost in estimates if cost is not None)
  345. def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse:
  346. """Add nested archive/printer/library_file info to response."""
  347. # Parse ams_mapping from JSON string BEFORE model_validate
  348. ams_mapping_parsed = None
  349. if item.ams_mapping:
  350. try:
  351. ams_mapping_parsed = json.loads(item.ams_mapping)
  352. except json.JSONDecodeError:
  353. ams_mapping_parsed = None
  354. # Parse required_filament_types from JSON string
  355. required_filament_types_parsed = None
  356. if item.required_filament_types:
  357. try:
  358. required_filament_types_parsed = json.loads(item.required_filament_types)
  359. except json.JSONDecodeError:
  360. required_filament_types_parsed = None
  361. # Parse filament_overrides from JSON string
  362. filament_overrides_parsed = None
  363. if item.filament_overrides:
  364. try:
  365. filament_overrides_parsed = json.loads(item.filament_overrides)
  366. except json.JSONDecodeError:
  367. filament_overrides_parsed = None
  368. # Parse nozzle_mapping from JSON string (#1780 — H2C rack slicer-pick
  369. # preservation). Nullable opaque JSON blob stored verbatim from
  370. # BambuStudio's project_file; surface it parsed for the response model
  371. # and any future "edit print → nozzle" UI.
  372. nozzle_mapping_parsed = None
  373. if item.nozzle_mapping:
  374. try:
  375. nozzle_mapping_parsed = json.loads(item.nozzle_mapping)
  376. except json.JSONDecodeError:
  377. nozzle_mapping_parsed = None
  378. # The operator's rack-position pick (#1784), keyed by filament group. Sent
  379. # parsed so the print dialog can show which hotend each group will use.
  380. nozzle_rack_choice_parsed = None
  381. if item.nozzle_rack_choice:
  382. try:
  383. nozzle_rack_choice_parsed = json.loads(item.nozzle_rack_choice)
  384. except json.JSONDecodeError:
  385. nozzle_rack_choice_parsed = None
  386. nozzles_info_parsed = None
  387. if item.nozzles_info:
  388. try:
  389. nozzles_info_parsed = json.loads(item.nozzles_info)
  390. except json.JSONDecodeError:
  391. nozzles_info_parsed = None
  392. # Create response with parsed ams_mapping
  393. item_dict = {
  394. "id": item.id,
  395. "printer_id": item.printer_id,
  396. "target_model": item.target_model,
  397. "target_location": item.target_location,
  398. "required_filament_types": required_filament_types_parsed,
  399. "filament_overrides": filament_overrides_parsed,
  400. "waiting_reason": item.waiting_reason,
  401. "archive_id": item.archive_id,
  402. "library_file_id": item.library_file_id,
  403. "cost_center_id": item.cost_center_id,
  404. "estimated_cost": item.estimated_cost,
  405. "position": item.position,
  406. "scheduled_time": item.scheduled_time,
  407. "require_previous_success": item.require_previous_success,
  408. "auto_off_after": item.auto_off_after,
  409. "manual_start": item.manual_start,
  410. "filament_short": bool(item.filament_short),
  411. "skip_filament_check": bool(item.skip_filament_check),
  412. "ams_mapping": ams_mapping_parsed,
  413. "plate_id": item.plate_id,
  414. "bed_levelling": item.bed_levelling,
  415. "flow_cali": item.flow_cali,
  416. "vibration_cali": item.vibration_cali,
  417. "layer_inspect": item.layer_inspect,
  418. "timelapse": item.timelapse,
  419. "use_ams": item.use_ams,
  420. "nozzle_offset_cali": item.nozzle_offset_cali,
  421. "preheat_override": item.preheat_override,
  422. "preheat_chamber_target_override": item.preheat_chamber_target_override,
  423. "status": item.status,
  424. "started_at": item.started_at,
  425. "completed_at": item.completed_at,
  426. "error_message": item.error_message,
  427. "created_at": item.created_at,
  428. # User tracking (Issue #206)
  429. "created_by_id": item.created_by_id,
  430. "created_by_username": item.created_by.username if item.created_by else None,
  431. # Batch grouping
  432. "batch_id": item.batch_id,
  433. "batch_name": item.batch.name if item.batch else None,
  434. # SJF scheduling
  435. "been_jumped": item.been_jumped,
  436. # Auto-print G-code injection
  437. "gcode_injection": item.gcode_injection,
  438. # H2C rack-swap nozzle pick (#1780)
  439. "nozzle_mapping": nozzle_mapping_parsed,
  440. "nozzle_rack_choice": nozzle_rack_choice_parsed,
  441. "nozzles_info": nozzles_info_parsed,
  442. "cleanup_library_after_dispatch": item.cleanup_library_after_dispatch,
  443. # Cross-model alternatives (#671). Guarded rather than read directly:
  444. # every route that reaches here eager-loads the relationship, but a
  445. # caller that forgets would trigger a lazy load, and a lazy load on an
  446. # async session raises rather than degrading. An empty list is the
  447. # correct answer for the ordinary item this would most likely be.
  448. "variants": _variant_summaries(item),
  449. }
  450. response = PrintQueueItemResponse(**item_dict)
  451. if item.archive:
  452. # Soft-deleted archive: files are gone from disk but the row stays
  453. # (its filament/cost contribution still flows into stats per #1343).
  454. # Suppress the archive-derived UI surface so the queue page doesn't
  455. # 404-storm the thumbnail / plates / plate-thumbnail endpoints — the
  456. # frontend's existing truthy gate on archive_thumbnail covers it
  457. # (#1348 follow-up). The archive_deleted flag lets the UI render a
  458. # "source deleted" badge on these rows.
  459. if item.archive.deleted_at is not None:
  460. response.archive_deleted = True
  461. else:
  462. response.archive_name = item.archive.print_name or item.archive.filename
  463. response.archive_thumbnail = item.archive.thumbnail_path
  464. response.print_time_seconds = item.archive.print_time_seconds
  465. response.filament_used_grams = item.archive.filament_used_grams
  466. response.filament_type = item.archive.filament_type
  467. response.filament_color = item.archive.filament_color
  468. response.layer_height = item.archive.layer_height
  469. response.nozzle_diameter = item.archive.nozzle_diameter
  470. response.sliced_for_model = item.archive.sliced_for_model
  471. response.bed_type = item.archive.bed_type
  472. # Marks history/reprint rows whose archive carries the slicer's own
  473. # live-resolved AMS-slot pick (extra_data.slicer_ams_mapping) — see
  474. # `_extract_slicer_ams_mapping_json` in virtual_printer/manager.py.
  475. #
  476. # Only when the saved mapping was resolved against *this* row's
  477. # printer: a global tray ID means nothing on another printer, so
  478. # that's the exact condition under which the mapping is reused. A
  479. # badge on a row where nothing gets reused would be a lie (#2700
  480. # review). Model-based rows (printer_id None) never match, which is
  481. # correct — the mapping is not reused there either.
  482. extra = item.archive.extra_data if isinstance(item.archive.extra_data, dict) else {}
  483. saved_mapping = extra.get("slicer_ams_mapping")
  484. response.archive_has_slicer_ams_mapping = (
  485. isinstance(saved_mapping, dict)
  486. and isinstance(saved_mapping.get("mapping"), list)
  487. and item.printer_id is not None
  488. and saved_mapping.get("printer_id") == item.printer_id
  489. )
  490. if item.plate_id:
  491. archive_path = settings.base_dir / item.archive.file_path
  492. if archive_path.exists():
  493. # One cached parse for all three per-plate overrides (#2573).
  494. plate_meta = extract_plate_metadata_from_3mf(archive_path, item.plate_id)
  495. if plate_meta.print_time_seconds is not None:
  496. response.print_time_seconds = plate_meta.print_time_seconds
  497. if plate_meta.filament_used_grams > 0:
  498. response.filament_used_grams = plate_meta.filament_used_grams
  499. if plate_meta.bed_type:
  500. response.bed_type = plate_meta.bed_type
  501. if item.library_file:
  502. response.library_file_name = (
  503. item.library_file.file_metadata.get("print_name") if item.library_file.file_metadata else None
  504. )
  505. if not response.library_file_name:
  506. response.library_file_name = item.library_file.filename
  507. response.library_file_thumbnail = item.library_file.thumbnail_path
  508. # Get metadata from library file if no archive
  509. if not item.archive and item.library_file.file_metadata:
  510. response.print_time_seconds = item.library_file.file_metadata.get("print_time_seconds")
  511. response.filament_used_grams = item.library_file.file_metadata.get("filament_used_grams")
  512. response.filament_type = item.library_file.file_metadata.get("filament_type")
  513. response.filament_color = item.library_file.file_metadata.get("filament_color")
  514. response.layer_height = item.library_file.file_metadata.get("layer_height")
  515. response.nozzle_diameter = item.library_file.file_metadata.get("nozzle_diameter")
  516. response.sliced_for_model = item.library_file.file_metadata.get("sliced_for_model")
  517. response.bed_type = item.library_file.file_metadata.get("bed_type")
  518. if item.plate_id:
  519. lib_path = Path(item.library_file.file_path)
  520. library_file_path = lib_path if lib_path.is_absolute() else settings.base_dir / item.library_file.file_path
  521. if library_file_path.exists():
  522. # One cached parse for all three per-plate overrides (#2573).
  523. plate_meta = extract_plate_metadata_from_3mf(library_file_path, item.plate_id)
  524. if plate_meta.print_time_seconds is not None:
  525. response.print_time_seconds = plate_meta.print_time_seconds
  526. if plate_meta.filament_used_grams > 0:
  527. response.filament_used_grams = plate_meta.filament_used_grams
  528. if plate_meta.bed_type:
  529. response.bed_type = plate_meta.bed_type
  530. if item.printer:
  531. response.printer_name = item.printer.name
  532. return response
  533. @router.get("/", response_model=list[PrintQueueItemResponse])
  534. async def list_queue(
  535. printer_id: int | None = Query(None, description="Filter by printer (-1 for unassigned)"),
  536. status: str | None = Query(None, description="Filter by status"),
  537. target_model: str | None = Query(
  538. None, description="Filter by target model (also includes model-based items when combined with printer_id)"
  539. ),
  540. db: AsyncSession = Depends(get_db),
  541. auth_result: tuple[User | None, bool] = Depends(
  542. require_ownership_permission(
  543. Permission.QUEUE_READ_ALL,
  544. Permission.QUEUE_READ_OWN,
  545. )
  546. ),
  547. ):
  548. """List all queue items, optionally filtered by printer or status."""
  549. user, can_read_all = auth_result
  550. query = (
  551. select(PrintQueueItem)
  552. .options(
  553. selectinload(PrintQueueItem.archive),
  554. selectinload(PrintQueueItem.printer),
  555. selectinload(PrintQueueItem.library_file),
  556. selectinload(PrintQueueItem.created_by),
  557. selectinload(PrintQueueItem.batch),
  558. # Cross-model candidates (#671) and their files, for the card label.
  559. selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
  560. )
  561. .order_by(PrintQueueItem.printer_id.nulls_first(), PrintQueueItem.position)
  562. )
  563. if user is not None and not can_read_all:
  564. query = query.where(PrintQueueItem.created_by_id == user.id)
  565. if printer_id is not None:
  566. if printer_id == -1:
  567. # Special value: filter for unassigned items
  568. query = query.where(PrintQueueItem.printer_id.is_(None))
  569. else:
  570. # Resolve effective model: prefer explicit param, fall back to printer's DB model.
  571. # This ensures model-based "Any X" items are returned even when the frontend
  572. # doesn't send target_model (e.g. printer.model is NULL on the client side).
  573. effective_model = target_model
  574. if not effective_model:
  575. printer_row = (
  576. await db.execute(select(Printer.model).where(Printer.id == printer_id))
  577. ).scalar_one_or_none()
  578. effective_model = printer_row
  579. if effective_model:
  580. # Include both printer-specific items AND model-based (unassigned) items
  581. query = query.where(
  582. or_(
  583. PrintQueueItem.printer_id == printer_id,
  584. and_(
  585. PrintQueueItem.printer_id.is_(None),
  586. func.lower(PrintQueueItem.target_model) == effective_model.lower(),
  587. ),
  588. )
  589. )
  590. else:
  591. query = query.where(PrintQueueItem.printer_id == printer_id)
  592. elif target_model:
  593. query = query.where(func.lower(PrintQueueItem.target_model) == target_model.lower())
  594. if status:
  595. query = query.where(PrintQueueItem.status == status)
  596. result = await db.execute(query)
  597. items = result.scalars().all()
  598. return [_enrich_response(item) for item in items]
  599. async def _resolve_queue_variants(
  600. db: AsyncSession,
  601. specs: list[QueueVariantCreate],
  602. current_user: User | None,
  603. ) -> list[tuple[QueueVariantCreate, LibraryFile, str]]:
  604. """Validate a cross-model candidate set and pair each file with its model (#671).
  605. Validated as a set, not file by file, because the failure modes are about the
  606. set: two candidates for the same printer give the resolver no basis to choose,
  607. and a set where nothing can ever run is a job that waits forever.
  608. At least one candidate must have an active printer — the rest may not, which
  609. is deliberate. Grouping the H2C slice before the H2C arrives is a reasonable
  610. thing to do, and refusing the whole queue action over it would be worse than
  611. letting that candidate simply never match.
  612. """
  613. file_ids = [s.library_file_id for s in specs]
  614. if len(set(file_ids)) != len(file_ids):
  615. raise HTTPException(400, "The same file cannot be listed twice as a variant")
  616. rows = (await db.execute(LibraryFile.active().where(LibraryFile.id.in_(file_ids)))).scalars().all()
  617. by_id = {f.id: f for f in rows}
  618. resolved: list[tuple[QueueVariantCreate, LibraryFile, str]] = []
  619. seen_models: dict[str, str] = {}
  620. any_active_printer = False
  621. for spec in specs:
  622. library_file = by_id.get(spec.library_file_id)
  623. # Same IDOR posture as the single-file path: a file the caller cannot read
  624. # is reported as missing rather than forbidden.
  625. if not library_file or (
  626. current_user
  627. and not current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
  628. and library_file.created_by_id != current_user.id
  629. ):
  630. raise HTTPException(404, f"Library file not found: {spec.library_file_id}")
  631. from backend.app.utils.filename import InvalidFilenameError, validate_print_filename
  632. try:
  633. validate_print_filename(library_file.filename)
  634. except InvalidFilenameError as e:
  635. raise HTTPException(400, str(e)) from e
  636. model = resolve_variant_model(library_file, spec.target_model)
  637. if not model:
  638. raise HTTPException(
  639. 400,
  640. f"{library_file.filename} does not say which printer it was sliced for — "
  641. "set its target model explicitly",
  642. )
  643. # Cross-model safety gate (#2578), per candidate. A set is only as safe as
  644. # its worst member, and model-based dispatch has no human in the loop.
  645. sliced_for = (library_file.file_metadata or {}).get("sliced_for_model")
  646. if not is_gcode_compatible(sliced_for, model):
  647. raise HTTPException(
  648. 400,
  649. f"{library_file.filename} was sliced for {sliced_for} and cannot be dispatched to {model} printers",
  650. )
  651. if model in seen_models:
  652. raise HTTPException(
  653. 400,
  654. f"{library_file.filename} and {seen_models[model]} are both for {model} — "
  655. "variants must target different printers",
  656. )
  657. seen_models[model] = library_file.filename
  658. has_printer = (
  659. (
  660. await db.execute(
  661. select(Printer).where(Printer.model == model).where(Printer.is_active == True) # noqa: E712
  662. )
  663. )
  664. .scalars()
  665. .first()
  666. )
  667. any_active_printer = any_active_printer or bool(has_printer)
  668. resolved.append((spec, library_file, model))
  669. if not any_active_printer:
  670. raise HTTPException(400, f"No active printers for any of: {', '.join(seen_models)}")
  671. return resolved
  672. def _variant_values(
  673. spec: QueueVariantCreate,
  674. library_file: LibraryFile,
  675. model: str,
  676. position: int,
  677. ) -> dict:
  678. """Column values for one candidate, extracted from its own 3MF.
  679. Each candidate is a different slice, so its filament requirements and print
  680. time come from its own file rather than being inherited from the item.
  681. Returns values rather than a row so a quantity>1 batch can build one row per
  682. copy without re-opening the 3MF for each.
  683. """
  684. lib_path = Path(library_file.file_path)
  685. file_path = lib_path if lib_path.is_absolute() else settings.base_dir / library_file.file_path
  686. required_types = None
  687. filament_overrides_json = None
  688. print_time = (library_file.file_metadata or {}).get("print_time_seconds")
  689. if file_path.exists():
  690. types = _extract_filament_types_from_3mf(file_path, spec.plate_id)
  691. if types:
  692. required_types = json.dumps(types)
  693. if spec.plate_id:
  694. plate_time = _extract_print_time_from_3mf(file_path, spec.plate_id)
  695. if plate_time is not None:
  696. print_time = plate_time
  697. if spec.filament_overrides:
  698. plate_overrides = overrides_for_plate(spec.filament_overrides, file_path, spec.plate_id)
  699. if plate_overrides:
  700. filament_overrides_json = json.dumps(plate_overrides)
  701. override_types = sorted({o["type"] for o in plate_overrides if "type" in o})
  702. if override_types:
  703. existing = set(json.loads(required_types)) if required_types else set()
  704. required_types = json.dumps(sorted(existing | set(override_types)))
  705. return {
  706. "position": position,
  707. "library_file_id": library_file.id,
  708. "target_model": model,
  709. "plate_id": spec.plate_id,
  710. "ams_mapping": json.dumps(spec.ams_mapping) if spec.ams_mapping else None,
  711. "nozzle_mapping": json.dumps(spec.nozzle_mapping) if spec.nozzle_mapping else None,
  712. "nozzle_rack_choice": json.dumps(spec.nozzle_rack_choice) if spec.nozzle_rack_choice else None,
  713. "filament_overrides": filament_overrides_json,
  714. "required_filament_types": required_types,
  715. "print_time_seconds": print_time,
  716. }
  717. @router.post("/", response_model=PrintQueueItemResponse)
  718. async def add_to_queue(
  719. data: PrintQueueItemCreate,
  720. db: AsyncSession = Depends(get_db),
  721. current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_CREATE),
  722. ):
  723. """Add an item to the print queue."""
  724. # Normalize target_model (e.g., "Bambu Lab X1E" / "C13" -> "X1E").
  725. # normalize_model_name resolves internal codes first: the previous
  726. # `normalize_printer_model(x) or normalize_printer_model_id(x)` chain never
  727. # reached the code map, because the first call returns unknown input
  728. # unchanged — so a "C13" target stayed "C13", matched no printer row and
  729. # left the item waiting forever. Identical result for every other spelling.
  730. target_model_norm = normalize_model_name(data.target_model)
  731. # Cross-model alternatives (#671): several sliced files, whichever printer
  732. # frees up first. The whole candidate set is validated before anything is
  733. # written — a half-valid set would produce a job that can only reach some of
  734. # the printers the user asked for, with nothing to say which.
  735. variant_specs: list[tuple[QueueVariantCreate, LibraryFile, str]] = []
  736. if data.variants:
  737. if data.printer_id:
  738. raise HTTPException(
  739. 400, "Cannot specify both printer_id and variants — pick a printer or offer alternatives"
  740. )
  741. if data.archive_id or data.library_file_id:
  742. raise HTTPException(
  743. 400, "Cannot combine variants with archive_id or library_file_id — the variants are the files"
  744. )
  745. variant_specs = await _resolve_queue_variants(db, data.variants, current_user)
  746. # Mirror the first candidate onto the item so the queue listing, the SJF
  747. # grouping and the "Any H2S" label have something before a printer is
  748. # picked. Resolution overwrites it with whichever candidate actually runs.
  749. target_model_norm = variant_specs[0][2]
  750. # Validate that either archive_id or library_file_id is provided.
  751. # A cross-model item deliberately holds neither: its files live on the
  752. # variant rows. Pointing library_file_id at one of them would be worse than
  753. # useless — that FK is ON DELETE CASCADE, so deleting a single alternative
  754. # would take the whole queue item with it.
  755. if not data.archive_id and not data.library_file_id and not data.variants:
  756. raise HTTPException(400, "Either archive_id or library_file_id must be provided")
  757. # Cannot specify both printer_id and target_model
  758. if data.printer_id and target_model_norm:
  759. raise HTTPException(400, "Cannot specify both printer_id and target_model")
  760. # Validate printer exists (if assigned)
  761. if data.printer_id is not None:
  762. result = await db.execute(select(Printer).where(Printer.id == data.printer_id))
  763. if not result.scalar_one_or_none():
  764. raise HTTPException(400, "Printer not found")
  765. # Validate target_model has active printers. Skipped for cross-model items:
  766. # target_model there is just the first candidate, and _resolve_queue_variants
  767. # has already required that *some* candidate has a printer.
  768. if target_model_norm and not data.variants:
  769. result = await db.execute(
  770. select(Printer).where(Printer.model == target_model_norm).where(Printer.is_active == True) # noqa: E712
  771. )
  772. if not result.scalars().first():
  773. raise HTTPException(400, f"No active printers for model: {target_model_norm}")
  774. # Validate archive exists (if provided) and get it for filament extraction
  775. archive = None
  776. if data.archive_id:
  777. result = await db.execute(select(PrintArchive).where(PrintArchive.id == data.archive_id))
  778. archive = result.scalar_one_or_none()
  779. if not archive:
  780. raise HTTPException(400, "Archive not found")
  781. _assert_can_queue_archive(archive, current_user)
  782. # Validate library file exists (if provided) and get it for filament extraction
  783. library_file = None
  784. if data.library_file_id:
  785. result = await db.execute(LibraryFile.active().where(LibraryFile.id == data.library_file_id))
  786. library_file = result.scalar_one_or_none()
  787. if not library_file:
  788. raise HTTPException(400, "Library file not found")
  789. _assert_can_queue_library_file(library_file, current_user)
  790. # Bambu SD card is FAT32/exFAT — illegal filename chars would 553 at
  791. # FTP upload time (#1540). Reject at queue time so the user gets the
  792. # actionable error before waiting in queue.
  793. from backend.app.utils.filename import InvalidFilenameError, validate_print_filename
  794. try:
  795. validate_print_filename(library_file.filename)
  796. except InvalidFilenameError as e:
  797. raise HTTPException(400, str(e)) from e
  798. # Cross-model safety gate (#2578): a G-code 3MF sliced for one model must
  799. # not be queued for dispatch to an incompatible model. The UI can no longer
  800. # produce such rows, but API-created rows must be rejected here too — the
  801. # scheduler assigns model-based items to hardware with no human in the loop.
  802. if target_model_norm:
  803. sliced_for = None
  804. if archive:
  805. sliced_for = archive.sliced_for_model
  806. elif library_file and library_file.file_metadata:
  807. sliced_for = library_file.file_metadata.get("sliced_for_model")
  808. if not is_gcode_compatible(sliced_for, target_model_norm):
  809. raise HTTPException(
  810. 400,
  811. f"File was sliced for {sliced_for} and cannot be dispatched to {target_model_norm} printers",
  812. )
  813. # Extract filament types for model-based assignment (used by scheduler for validation)
  814. required_filament_types = None
  815. file_path = None
  816. if target_model_norm:
  817. # Get file path from archive or library file
  818. if archive:
  819. file_path = settings.base_dir / archive.file_path
  820. elif library_file:
  821. lib_path = Path(library_file.file_path)
  822. file_path = lib_path if lib_path.is_absolute() else settings.base_dir / library_file.file_path
  823. if file_path and file_path.exists():
  824. filament_types = _extract_filament_types_from_3mf(file_path, data.plate_id)
  825. if filament_types:
  826. required_filament_types = json.dumps(filament_types)
  827. logger.info("Extracted filament types for model-based queue: %s", filament_types)
  828. # If filament overrides are provided, update required_filament_types to match override types
  829. filament_overrides_json = None
  830. if data.filament_overrides and target_model_norm:
  831. plate_overrides = overrides_for_plate(data.filament_overrides, file_path, data.plate_id)
  832. if plate_overrides:
  833. filament_overrides_json = json.dumps(plate_overrides)
  834. # Update required_filament_types from overrides so scheduler validates against overridden types
  835. override_types = sorted({o["type"] for o in plate_overrides if "type" in o})
  836. if override_types:
  837. # Merge with existing types (overrides may only cover some slots)
  838. existing_types = set(json.loads(required_filament_types)) if required_filament_types else set()
  839. # Replace types for overridden slots, keep others
  840. all_types = existing_types | set(override_types)
  841. required_filament_types = json.dumps(sorted(all_types))
  842. # Validate quantity
  843. quantity = max(1, data.quantity)
  844. # Validate batch_id if provided. Client passes batch_id when adding items
  845. # into a pre-created batch (multi-plate auto-batch or "Group as batch" flow).
  846. # 404 keeps the existing-id leak surface low.
  847. batch = None
  848. batch_id = None
  849. if data.batch_id is not None:
  850. result = await db.execute(select(PrintBatch).where(PrintBatch.id == data.batch_id))
  851. existing_batch = result.scalar_one_or_none()
  852. if not existing_batch:
  853. raise HTTPException(404, "Batch not found")
  854. if existing_batch.status != "active":
  855. raise HTTPException(400, "Cannot add items to a non-active batch")
  856. if (
  857. current_user is not None
  858. and existing_batch.created_by_id is not None
  859. and existing_batch.created_by_id != current_user.id
  860. and not current_user.has_permission(Permission.QUEUE_UPDATE_ALL.value)
  861. ):
  862. raise HTTPException(404, "Batch not found")
  863. batch = existing_batch
  864. batch_id = existing_batch.id
  865. # Create batch if quantity > 1 and no batch_id provided
  866. if batch_id is None and quantity > 1:
  867. # Derive batch name from source file
  868. batch_name_base = "Batch"
  869. if archive:
  870. batch_name_base = archive.print_name or archive.filename or "Batch"
  871. elif library_file:
  872. if library_file.file_metadata:
  873. batch_name_base = library_file.file_metadata.get("print_name") or library_file.filename
  874. else:
  875. batch_name_base = library_file.filename
  876. batch_name_base = batch_name_base.replace(".gcode.3mf", "").replace(".3mf", "")
  877. batch = PrintBatch(
  878. name=f"{batch_name_base} ×{quantity}",
  879. archive_id=data.archive_id,
  880. library_file_id=data.library_file_id,
  881. quantity=quantity,
  882. status="active",
  883. created_by_id=current_user.id if current_user else None,
  884. )
  885. db.add(batch)
  886. await db.flush() # Get batch.id before creating items
  887. batch_id = batch.id
  888. # Get queue scope for this printer (or for unassigned/model-based items).
  889. if data.printer_id is not None:
  890. queue_scope = (
  891. PrintQueueItem.printer_id == data.printer_id,
  892. PrintQueueItem.status == "pending",
  893. )
  894. else:
  895. # For unassigned/model-based items, scope across all unassigned.
  896. queue_scope = (
  897. PrintQueueItem.printer_id.is_(None),
  898. PrintQueueItem.status == "pending",
  899. )
  900. # Serialize concurrent queue inserts to the same scope (#1625-followup).
  901. # The race: two concurrent ASAP inserts both compute MAX(position) before
  902. # either commits; in an empty scope, both INSERT at position 1 (duplicate).
  903. # In a non-empty scope, Postgres's row-level locks on the UPDATE shift
  904. # serialize naturally, but the empty-scope path has no rows to lock.
  905. # A transaction-scoped advisory lock keyed on the printer_id closes that
  906. # window; the lock is released automatically at commit/rollback. Different
  907. # printers don't contend. SQLite serializes writes implicitly so this is a
  908. # no-op there.
  909. #
  910. # Dialect is checked against the actual session binding, NOT the
  911. # `is_sqlite()` helper, because the test fixture overrides `get_db` with a
  912. # SQLite engine while `settings.database_url` still points at Postgres
  913. # (the helper reads settings). Inspecting the connection directly is the
  914. # right shape for any code that mutates SQL based on the live dialect.
  915. from sqlalchemy import text
  916. bind = db.get_bind()
  917. if bind.dialect.name == "postgresql":
  918. scope_key = data.printer_id if data.printer_id is not None else 0
  919. # 1625 namespaces the lock so it can't collide with other advisory
  920. # locks elsewhere in the codebase.
  921. await db.execute(text("SELECT pg_advisory_xact_lock(1625, :k)"), {"k": scope_key})
  922. insert_position = max(1, data.insert_position or 1)
  923. if data.insert_at_top or data.insert_position is not None:
  924. result = await db.execute(select(func.max(PrintQueueItem.position)).where(*queue_scope))
  925. max_pos = result.scalar() or 0
  926. insert_position = min(insert_position, max_pos + 1)
  927. await db.execute(
  928. update(PrintQueueItem)
  929. .where(*queue_scope)
  930. .where(PrintQueueItem.position >= insert_position)
  931. .values(position=PrintQueueItem.position + quantity)
  932. )
  933. start_position = insert_position
  934. else:
  935. result = await db.execute(select(func.max(PrintQueueItem.position)).where(*queue_scope))
  936. max_pos = result.scalar() or 0
  937. start_position = max_pos + 1
  938. # Resolve print_time_seconds for SJF scheduling (cache on item at creation)
  939. cached_print_time = None
  940. if archive:
  941. cached_print_time = archive.print_time_seconds
  942. if data.plate_id:
  943. archive_path = settings.base_dir / archive.file_path
  944. if archive_path.exists():
  945. plate_time = _extract_print_time_from_3mf(archive_path, data.plate_id)
  946. if plate_time is not None:
  947. cached_print_time = plate_time
  948. elif library_file:
  949. if library_file.file_metadata:
  950. cached_print_time = library_file.file_metadata.get("print_time_seconds")
  951. if data.plate_id:
  952. lib_path = Path(library_file.file_path)
  953. library_file_path = lib_path if lib_path.is_absolute() else settings.base_dir / library_file.file_path
  954. if library_file_path.exists():
  955. plate_time = _extract_print_time_from_3mf(library_file_path, data.plate_id)
  956. if plate_time is not None:
  957. cached_print_time = plate_time
  958. # Validate project exists before insert so a bogus ID yields 404, not an FK-constraint 500
  959. if data.project_id is not None:
  960. project_result = await db.execute(select(Project).where(Project.id == data.project_id))
  961. if not project_result.scalar_one_or_none():
  962. raise HTTPException(status_code=404, detail="Project not found")
  963. # Security boundary: the browser's estimated_cost is only a display hint.
  964. # Budget enforcement and the persisted reservation value must be derived
  965. # from the server-owned archive/library metadata and spool assignments.
  966. if variant_specs:
  967. variant_costs = [
  968. await estimate_queue_source_cost(
  969. db,
  970. library_file=variant_file,
  971. plate_id=spec.plate_id,
  972. ams_mapping=spec.ams_mapping,
  973. printer_id=data.printer_id,
  974. )
  975. for spec, variant_file, _model in variant_specs
  976. ]
  977. trusted_estimated_cost = (
  978. max(cost for cost in variant_costs if cost is not None)
  979. if variant_costs and all(cost is not None for cost in variant_costs)
  980. else None
  981. )
  982. else:
  983. trusted_estimated_cost = await estimate_queue_source_cost(
  984. db,
  985. archive=archive,
  986. library_file=library_file,
  987. plate_id=data.plate_id,
  988. ams_mapping=data.ams_mapping,
  989. printer_id=data.printer_id,
  990. )
  991. await validate_print_budget(
  992. db,
  993. cost_center_id=data.cost_center_id,
  994. estimated_cost=trusted_estimated_cost,
  995. current_user=current_user,
  996. quantity=quantity,
  997. )
  998. ams_mapping_json = json.dumps(data.ams_mapping) if data.ams_mapping else None
  999. # Same Text-as-JSON convention for the rack-position pick (#1784).
  1000. nozzle_rack_choice_json = json.dumps(data.nozzle_rack_choice) if data.nozzle_rack_choice else None
  1001. # Reprint fallback: the caller didn't specify an explicit ams_mapping (no
  1002. # per-slot filament-mapping edit was made), but the archive carries the
  1003. # slicer's own live-resolved AMS-slot pick from the original print (see
  1004. # `extra_data.slicer_ams_mapping`, written by the VP-queue path via
  1005. # `_extract_slicer_ams_mapping_json`). Reuse it so the reprint dispatches
  1006. # to the exact same physical spool instead of the scheduler re-deriving a
  1007. # (possibly ambiguous) mapping from just the file's static type/color.
  1008. #
  1009. # Global tray IDs only mean something relative to the specific printer
  1010. # they were resolved against, so this only fires when the reprint targets
  1011. # that exact printer (`extra_data.slicer_ams_mapping.printer_id`) — never
  1012. # for a model-based dispatch (data.printer_id is None) or a reprint aimed
  1013. # at a different printer, where the same tray number can hold a
  1014. # completely different spool (#2700 review).
  1015. #
  1016. # It also stands down when the request carries force-color-match overrides:
  1017. # those are the caller asking the scheduler to match strictly against the
  1018. # printer's live trays, and they are only ever applied inside
  1019. # `_compute_ams_mapping_for_printer` — the function a stored mapping makes
  1020. # the scheduler skip. Same precedence as the VP-side toggle pair (#2700
  1021. # review).
  1022. #
  1023. # Note this is otherwise unconditional — it applies regardless of whether
  1024. # the physical spool in that slot has changed since the original print.
  1025. # #1308 covers re-verifying a stored mapping against live AMS state at
  1026. # dispatch time; that check is a separate PR and, once merged, will also
  1027. # catch a stale slot inherited through this fallback.
  1028. wants_live_color_match = any(
  1029. isinstance(o, dict) and o.get("force_color_match") for o in (data.filament_overrides or [])
  1030. )
  1031. if (
  1032. ams_mapping_json is None
  1033. and not wants_live_color_match
  1034. and archive
  1035. and archive.extra_data
  1036. and data.printer_id is not None
  1037. ):
  1038. saved = archive.extra_data.get("slicer_ams_mapping")
  1039. if (
  1040. isinstance(saved, dict)
  1041. and saved.get("printer_id") == data.printer_id
  1042. and isinstance(saved.get("mapping"), list)
  1043. and saved["mapping"]
  1044. ):
  1045. ams_mapping_json = json.dumps(saved["mapping"])
  1046. items = []
  1047. for i in range(quantity):
  1048. item = PrintQueueItem(
  1049. printer_id=data.printer_id,
  1050. target_model=target_model_norm,
  1051. target_location=data.target_location,
  1052. required_filament_types=required_filament_types,
  1053. filament_overrides=filament_overrides_json,
  1054. archive_id=data.archive_id,
  1055. library_file_id=data.library_file_id,
  1056. cost_center_id=data.cost_center_id,
  1057. estimated_cost=trusted_estimated_cost,
  1058. scheduled_time=data.scheduled_time,
  1059. require_previous_success=data.require_previous_success,
  1060. auto_off_after=data.auto_off_after,
  1061. manual_start=data.manual_start,
  1062. skip_filament_check=data.skip_filament_check,
  1063. ams_mapping=ams_mapping_json,
  1064. nozzle_rack_choice=nozzle_rack_choice_json,
  1065. plate_id=data.plate_id,
  1066. bed_levelling=data.bed_levelling,
  1067. flow_cali=data.flow_cali,
  1068. vibration_cali=data.vibration_cali,
  1069. layer_inspect=data.layer_inspect,
  1070. timelapse=data.timelapse,
  1071. use_ams=data.use_ams,
  1072. nozzle_offset_cali=data.nozzle_offset_cali,
  1073. preheat_override=data.preheat_override,
  1074. preheat_chamber_target_override=data.preheat_chamber_target_override,
  1075. gcode_injection=data.gcode_injection,
  1076. cleanup_library_after_dispatch=data.cleanup_library_after_dispatch,
  1077. project_id=data.project_id,
  1078. position=start_position + i,
  1079. status="pending",
  1080. created_by_id=current_user.id if current_user else None,
  1081. batch_id=batch_id,
  1082. print_time_seconds=cached_print_time,
  1083. )
  1084. db.add(item)
  1085. items.append(item)
  1086. if variant_specs:
  1087. variant_values = [
  1088. _variant_values(spec, library_file, model, position)
  1089. for position, (spec, library_file, model) in enumerate(variant_specs)
  1090. ]
  1091. # SJF orders pending items before any printer is known, so the row carries
  1092. # the shortest candidate's estimate. Resolution replaces it with the one
  1093. # that actually runs.
  1094. estimates = [v["print_time_seconds"] for v in variant_values if v["print_time_seconds"]]
  1095. for item in items:
  1096. # Each copy in a quantity>1 batch gets its own candidate rows —
  1097. # attempt counts are per-item, and two copies must be free to land on
  1098. # different printers.
  1099. item.variants.extend(PrintQueueVariant(**values) for values in variant_values)
  1100. item.print_time_seconds = min(estimates) if estimates else None
  1101. await db.commit()
  1102. # Refresh the first item for the response
  1103. item = items[0]
  1104. await db.refresh(item)
  1105. await db.refresh(item, ["archive", "printer", "library_file", "created_by", "batch"])
  1106. source_name = f"archive {data.archive_id}" if data.archive_id else f"library file {data.library_file_id}"
  1107. target_desc = data.printer_id or (f"model {target_model_norm}" if target_model_norm else "unassigned")
  1108. qty_desc = f" (×{quantity})" if quantity > 1 else ""
  1109. logger.info("Added %s to queue for %s%s", source_name, target_desc, qty_desc)
  1110. # MQTT relay - publish queue job added
  1111. try:
  1112. from backend.app.services.mqtt_relay import mqtt_relay
  1113. await mqtt_relay.on_queue_job_added(
  1114. job_id=item.id,
  1115. filename=item.archive.filename if item.archive else "",
  1116. printer_id=item.printer_id,
  1117. printer_name=item.printer.name if item.printer else None,
  1118. )
  1119. except Exception:
  1120. pass # Don't fail queue add if MQTT fails
  1121. # Send notification for job added
  1122. try:
  1123. job_name = (
  1124. item.archive.filename
  1125. if item.archive
  1126. else item.library_file.filename
  1127. if item.library_file
  1128. else f"Job #{item.id}"
  1129. )
  1130. job_name = job_name.replace(".gcode.3mf", "").replace(".3mf", "")
  1131. if quantity > 1:
  1132. job_name = f"{job_name} ×{quantity}"
  1133. target = (
  1134. item.printer.name if item.printer else (f"Any {item.target_model}" if target_model_norm else "Unassigned")
  1135. )
  1136. await notification_service.on_queue_job_added(
  1137. job_name=job_name,
  1138. target=target,
  1139. db=db,
  1140. printer_id=item.printer_id,
  1141. printer_name=item.printer.name if item.printer else None,
  1142. )
  1143. except Exception:
  1144. pass # Don't fail queue add if notification fails
  1145. return _enrich_response(item)
  1146. @router.patch("/bulk", response_model=PrintQueueBulkUpdateResponse)
  1147. async def bulk_update_queue_items(
  1148. data: PrintQueueBulkUpdate,
  1149. db: AsyncSession = Depends(get_db),
  1150. auth_result: tuple[User | None, bool] = Depends(
  1151. require_ownership_permission(
  1152. Permission.QUEUE_UPDATE_ALL,
  1153. Permission.QUEUE_UPDATE_OWN,
  1154. )
  1155. ),
  1156. ):
  1157. """Bulk update multiple queue items with the same values.
  1158. Only pending items can be updated. Non-pending items are skipped.
  1159. Items not owned by the user are also skipped (unless user has *_all permission).
  1160. """
  1161. user, can_modify_all = auth_result
  1162. if not data.item_ids:
  1163. raise HTTPException(400, "No item IDs provided")
  1164. # Get fields to update (exclude item_ids and unset fields)
  1165. update_data = data.model_dump(exclude={"item_ids"}, exclude_unset=True)
  1166. if not update_data:
  1167. raise HTTPException(400, "No fields to update")
  1168. # Validate printer_id if being changed
  1169. if "printer_id" in update_data and update_data["printer_id"] is not None:
  1170. result = await db.execute(select(Printer).where(Printer.id == update_data["printer_id"]))
  1171. if not result.scalar_one_or_none():
  1172. raise HTTPException(400, "Printer not found")
  1173. # Fetch all items
  1174. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id.in_(data.item_ids)))
  1175. items = result.scalars().all()
  1176. updated_count = 0
  1177. skipped_count = 0
  1178. validates_billing_fields = "cost_center_id" in update_data or "estimated_cost" in update_data
  1179. for item in items:
  1180. # Skip non-pending rows and rows a dispatch worker has claimed (#2615) —
  1181. # editing a claimed row mid-upload would split it from the in-flight
  1182. # dispatch, so it's excluded from the bulk change (cancel to move it).
  1183. if item.status != "pending" or item.dispatching_at is not None:
  1184. skipped_count += 1
  1185. continue
  1186. # Ownership check
  1187. if not can_modify_all and item.created_by_id != user.id:
  1188. skipped_count += 1
  1189. continue
  1190. item_update_data = update_data.copy()
  1191. if validates_billing_fields:
  1192. trusted_estimated_cost = await _trusted_item_estimated_cost(
  1193. db,
  1194. item,
  1195. printer_id=item_update_data.get("printer_id", item.printer_id),
  1196. plate_id=item.plate_id,
  1197. ams_mapping=item.ams_mapping,
  1198. )
  1199. item_update_data["estimated_cost"] = trusted_estimated_cost
  1200. await validate_print_budget(
  1201. db,
  1202. cost_center_id=item_update_data.get("cost_center_id", item.cost_center_id),
  1203. estimated_cost=trusted_estimated_cost,
  1204. current_user=user,
  1205. exclude_queue_item_id=item.id,
  1206. )
  1207. for field, value in item_update_data.items():
  1208. setattr(item, field, value)
  1209. updated_count += 1
  1210. await db.commit()
  1211. logger.info("Bulk updated %s queue items, skipped %s", updated_count, skipped_count)
  1212. return PrintQueueBulkUpdateResponse(
  1213. updated_count=updated_count,
  1214. skipped_count=skipped_count,
  1215. message=f"Updated {updated_count} items"
  1216. + (f", skipped {skipped_count} non-pending/not-owned" if skipped_count else ""),
  1217. )
  1218. # --- Batch endpoints ---
  1219. def _validate_plate_targets(
  1220. plates: list[PrintBatchPlateTarget] | None,
  1221. ) -> list[PrintBatchPlateTarget] | None:
  1222. """Reject duplicate plates and orders that ask for nothing at all.
  1223. A duplicate would violate the (batch_id, plate_id) unique constraint at
  1224. flush time — and on SQLite/PostgreSQL a NULL plate_id slips past that
  1225. constraint entirely, so the check has to happen here to catch two
  1226. "whole file" rows in one order.
  1227. """
  1228. if plates is None:
  1229. return None
  1230. if not plates:
  1231. raise HTTPException(400, "plates must contain at least one plate")
  1232. seen: set[int | None] = set()
  1233. for target in plates:
  1234. if target.plate_id in seen:
  1235. label = target.plate_id if target.plate_id is not None else "whole file"
  1236. raise HTTPException(400, f"Duplicate plate in order: {label}")
  1237. seen.add(target.plate_id)
  1238. if all(target.quantity_target == 0 for target in plates):
  1239. raise HTTPException(400, "Order must request at least one print")
  1240. return plates
  1241. async def _validate_batch_project(db: AsyncSession, project_id: int | None, current_user: User | None) -> None:
  1242. """404 on a bogus project id rather than letting the FK blow up as a 500."""
  1243. if project_id is None:
  1244. return
  1245. result = await db.execute(select(Project).where(Project.id == project_id))
  1246. if not result.scalar_one_or_none():
  1247. raise HTTPException(404, "Project not found")
  1248. async def _load_batch_for_write(
  1249. db: AsyncSession, batch_id: int, current_user: User | None, permission: Permission
  1250. ) -> PrintBatch:
  1251. """Fetch a batch the caller is allowed to modify, or 404.
  1252. 404 rather than 403 on the ownership miss, matching the rest of this
  1253. module: a 403 would confirm the id exists to someone enumerating.
  1254. """
  1255. result = await db.execute(
  1256. select(PrintBatch).options(selectinload(PrintBatch.plates)).where(PrintBatch.id == batch_id)
  1257. )
  1258. batch = result.scalar_one_or_none()
  1259. if not batch:
  1260. raise HTTPException(404, "Batch not found")
  1261. if (
  1262. current_user is not None
  1263. and batch.created_by_id is not None
  1264. and batch.created_by_id != current_user.id
  1265. and not current_user.has_permission(permission.value)
  1266. ):
  1267. raise HTTPException(404, "Batch not found")
  1268. return batch
  1269. @router.post("/batches", response_model=PrintBatchResponse)
  1270. async def create_batch(
  1271. data: PrintBatchCreate,
  1272. db: AsyncSession = Depends(get_db),
  1273. current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_CREATE),
  1274. ):
  1275. """Create a batch.
  1276. Two modes:
  1277. * ``item_ids`` provided: assign the listed pending queue items to a new
  1278. batch ("Group as batch" UI action).
  1279. * ``item_ids`` omitted/empty: create an empty batch so the client can
  1280. pass the returned ``id`` on subsequent ``POST /queue/`` calls. Used by
  1281. the multi-plate auto-batch flow in PrintModal.
  1282. ``plates`` turns the batch into an order with per-plate targets (#342):
  1283. progress is then measured against what was asked for rather than against
  1284. what happened to be queued, so a failed run still counts as owed. Omitting
  1285. it keeps the pre-#342 behaviour exactly.
  1286. """
  1287. if not data.name or not data.name.strip():
  1288. raise HTTPException(400, "Batch name is required")
  1289. plate_targets = _validate_plate_targets(data.plates)
  1290. await _validate_batch_project(db, data.project_id, current_user)
  1291. batch = PrintBatch(
  1292. name=data.name.strip()[:255],
  1293. archive_id=data.archive_id,
  1294. library_file_id=data.library_file_id,
  1295. quantity=len(data.item_ids) if data.item_ids else 1,
  1296. status="active",
  1297. created_by_id=current_user.id if current_user else None,
  1298. project_id=data.project_id,
  1299. due_date=data.due_date,
  1300. notes=data.notes,
  1301. )
  1302. db.add(batch)
  1303. await db.flush() # Need batch.id before assigning to items
  1304. if plate_targets is not None:
  1305. for target in plate_targets:
  1306. db.add(
  1307. PrintBatchPlate(
  1308. batch_id=batch.id,
  1309. plate_id=target.plate_id,
  1310. plate_name=target.plate_name,
  1311. quantity_target=target.quantity_target,
  1312. sort_order=target.sort_order,
  1313. )
  1314. )
  1315. # The legacy `quantity` column is display-only; keep it meaningful for
  1316. # anything still reading it by making it the order's total.
  1317. batch.quantity = max(1, sum(t.quantity_target for t in plate_targets))
  1318. assigned = 0
  1319. if data.item_ids:
  1320. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id.in_(data.item_ids)))
  1321. items = result.scalars().all()
  1322. for item in items:
  1323. if item.status != "pending":
  1324. continue
  1325. if item.batch_id is not None:
  1326. continue
  1327. if (
  1328. current_user is not None
  1329. and item.created_by_id != current_user.id
  1330. and not current_user.has_permission(Permission.QUEUE_UPDATE_ALL.value)
  1331. ):
  1332. continue
  1333. item.batch_id = batch.id
  1334. assigned += 1
  1335. batch.quantity = max(assigned, 1)
  1336. await db.commit()
  1337. await db.refresh(batch)
  1338. logger.info("Created batch %s '%s' with %s assigned items", batch.id, batch.name, assigned)
  1339. return await _build_batch_response(db, batch)
  1340. @router.patch("/batches/{batch_id}", response_model=PrintBatchResponse)
  1341. async def update_batch(
  1342. batch_id: int,
  1343. data: PrintBatchUpdate,
  1344. db: AsyncSession = Depends(get_db),
  1345. current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN),
  1346. ):
  1347. """Edit an order's header or its per-plate targets while it runs (#342).
  1348. Production requirements change mid-run, so targets are editable. Lowering a
  1349. target below what has already been dispatched is allowed and simply leaves
  1350. ``remaining`` at zero — cancelling the surplus queue items is a separate,
  1351. explicit action, because silently deleting queued work on a number change
  1352. would be a nasty surprise.
  1353. """
  1354. batch = await _load_batch_for_write(db, batch_id, current_user, Permission.QUEUE_UPDATE_ALL)
  1355. plate_targets = _validate_plate_targets(data.plates)
  1356. if data.project_id is not None:
  1357. await _validate_batch_project(db, data.project_id, current_user)
  1358. if data.name is not None:
  1359. if not data.name.strip():
  1360. raise HTTPException(400, "Batch name is required")
  1361. batch.name = data.name.strip()[:255]
  1362. if data.project_id is not None:
  1363. batch.project_id = data.project_id
  1364. if data.due_date is not None:
  1365. batch.due_date = data.due_date
  1366. if data.notes is not None:
  1367. batch.notes = data.notes
  1368. if data.status is not None:
  1369. batch.status = data.status
  1370. if plate_targets is not None:
  1371. existing = {row.plate_id: row for row in batch.plates}
  1372. for target in plate_targets:
  1373. row = existing.pop(target.plate_id, None)
  1374. if row is None:
  1375. db.add(
  1376. PrintBatchPlate(
  1377. batch_id=batch.id,
  1378. plate_id=target.plate_id,
  1379. plate_name=target.plate_name,
  1380. quantity_target=target.quantity_target,
  1381. sort_order=target.sort_order,
  1382. )
  1383. )
  1384. else:
  1385. row.quantity_target = target.quantity_target
  1386. row.sort_order = target.sort_order
  1387. if target.plate_name is not None:
  1388. row.plate_name = target.plate_name
  1389. # Plates absent from the payload are dropped — the list is the order.
  1390. for orphan in existing.values():
  1391. await db.delete(orphan)
  1392. batch.quantity = max(1, sum(t.quantity_target for t in plate_targets))
  1393. await db.flush()
  1394. await db.refresh(batch)
  1395. # Raising a target on a finished order reopens it; lowering one on a
  1396. # running order can complete it.
  1397. await refresh_batch_status(db, batch)
  1398. await db.commit()
  1399. await db.refresh(batch)
  1400. logger.info("Updated batch %s", batch.id)
  1401. return await _build_batch_response(db, batch)
  1402. @router.post("/batches/{batch_id}/dispatch", response_model=PrintBatchResponse)
  1403. async def dispatch_batch(
  1404. batch_id: int,
  1405. data: PrintBatchDispatchRequest,
  1406. db: AsyncSession = Depends(get_db),
  1407. current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_CREATE),
  1408. ):
  1409. """Queue the runs this order still owes (#342).
  1410. Each new item is cloned from the most recent item for the same plate in
  1411. this batch, so it inherits the printer/model target, AMS mapping, filament
  1412. overrides and print options the user already chose — and the validation
  1413. those went through at creation time.
  1414. """
  1415. batch = await _load_batch_for_write(db, batch_id, current_user, Permission.QUEUE_UPDATE_ALL)
  1416. if batch.status == "cancelled":
  1417. raise HTTPException(400, "Cannot dispatch a cancelled batch")
  1418. # Dispatch starts prints, so it must not be a weaker door than POST /queue/.
  1419. await _assert_can_dispatch_batch_sources(db, batch.id, current_user)
  1420. try:
  1421. created = await dispatch_remaining(
  1422. db,
  1423. batch,
  1424. plate_id=data.plate_id,
  1425. only_plate=data.only_plate,
  1426. limit=data.limit,
  1427. created_by_id=current_user.id if current_user else None,
  1428. )
  1429. except BatchDispatchError as exc:
  1430. raise HTTPException(400, str(exc)) from exc
  1431. await db.commit()
  1432. await db.refresh(batch)
  1433. logger.info("Batch %s dispatched %d item(s)", batch.id, len(created))
  1434. return await _build_batch_response(db, batch)
  1435. @router.post("/batches/{batch_id}/ungroup", response_model=PrintBatchUngroupResponse)
  1436. async def ungroup_batch(
  1437. batch_id: int,
  1438. db: AsyncSession = Depends(get_db),
  1439. current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN),
  1440. ):
  1441. """Disband a batch: clear batch_id from all members and delete the batch row.
  1442. Items stay in the queue. Only ungroups items the caller owns (unless they
  1443. hold QUEUE_UPDATE_ALL). A batch with all members ungrouped is deleted.
  1444. """
  1445. result = await db.execute(select(PrintBatch).where(PrintBatch.id == batch_id))
  1446. batch = result.scalar_one_or_none()
  1447. if not batch:
  1448. raise HTTPException(404, "Batch not found")
  1449. can_modify_all = current_user is None or current_user.has_permission(Permission.QUEUE_UPDATE_ALL.value)
  1450. if not can_modify_all and batch.created_by_id != (current_user.id if current_user else None):
  1451. raise HTTPException(404, "Batch not found")
  1452. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.batch_id == batch_id))
  1453. items = result.scalars().all()
  1454. ungrouped = 0
  1455. remaining = 0
  1456. for item in items:
  1457. if not can_modify_all and item.created_by_id != (current_user.id if current_user else None):
  1458. remaining += 1
  1459. continue
  1460. item.batch_id = None
  1461. ungrouped += 1
  1462. # Delete the batch row only when all members were ungrouped — otherwise it
  1463. # still owns the items the caller couldn't touch.
  1464. if remaining == 0:
  1465. await db.delete(batch)
  1466. await db.commit()
  1467. logger.info("Ungrouped batch %s (%s items)", batch_id, ungrouped)
  1468. return PrintBatchUngroupResponse(
  1469. ungrouped_count=ungrouped,
  1470. message=f"Ungrouped {ungrouped} item(s)",
  1471. )
  1472. @router.get("/batches", response_model=list[PrintBatchResponse])
  1473. async def list_batches(
  1474. status: str | None = Query(None, description="Filter by status (active, completed, cancelled)"),
  1475. db: AsyncSession = Depends(get_db),
  1476. auth_result: tuple[User | None, bool] = Depends(
  1477. require_ownership_permission(
  1478. Permission.QUEUE_READ_ALL,
  1479. Permission.QUEUE_READ_OWN,
  1480. )
  1481. ),
  1482. ):
  1483. """List print batches with progress stats.
  1484. Batches with neither queue items nor per-plate targets are omitted. Those
  1485. are empty shells — a grouping whose items were deleted with their source
  1486. archive, or a create that never got as far as adding any — and they carry
  1487. nothing to show, track or dispatch. A brand-new order is still listed
  1488. before its first dispatch, because its targets say what it owes.
  1489. """
  1490. current_user, can_read_all = auth_result
  1491. query = (
  1492. select(PrintBatch)
  1493. .where(
  1494. select(PrintQueueItem.id).where(PrintQueueItem.batch_id == PrintBatch.id).exists()
  1495. | select(PrintBatchPlate.id).where(PrintBatchPlate.batch_id == PrintBatch.id).exists()
  1496. )
  1497. .order_by(PrintBatch.created_at.desc())
  1498. )
  1499. if status:
  1500. query = query.where(PrintBatch.status == status)
  1501. if current_user is not None and not can_read_all:
  1502. query = query.where(PrintBatch.created_by_id == current_user.id)
  1503. result = await db.execute(query)
  1504. batches = result.scalars().all()
  1505. # Resolve creator names in one query rather than one per batch.
  1506. creator_ids = {b.created_by_id for b in batches if b.created_by_id is not None}
  1507. usernames: dict[int, str] = {}
  1508. if creator_ids:
  1509. rows = await db.execute(select(User.id, User.username).where(User.id.in_(creator_ids)))
  1510. usernames = {row[0]: row[1] for row in rows.all()}
  1511. return [await _build_batch_response(db, batch, usernames=usernames) for batch in batches]
  1512. @router.get("/batches/{batch_id}", response_model=PrintBatchResponse)
  1513. async def get_batch(
  1514. batch_id: int,
  1515. db: AsyncSession = Depends(get_db),
  1516. auth_result: tuple[User | None, bool] = Depends(
  1517. require_ownership_permission(
  1518. Permission.QUEUE_READ_ALL,
  1519. Permission.QUEUE_READ_OWN,
  1520. )
  1521. ),
  1522. ):
  1523. """Get a print batch with progress stats."""
  1524. current_user, can_read_all = auth_result
  1525. result = await db.execute(select(PrintBatch).where(PrintBatch.id == batch_id))
  1526. batch = result.scalar_one_or_none()
  1527. if not batch:
  1528. raise HTTPException(404, "Batch not found")
  1529. if (
  1530. current_user is not None
  1531. and not can_read_all
  1532. and (batch.created_by_id is None or batch.created_by_id != current_user.id)
  1533. ):
  1534. raise HTTPException(404, "Batch not found")
  1535. return await _build_batch_response(db, batch)
  1536. @router.delete("/batches/{batch_id}")
  1537. async def cancel_batch(
  1538. batch_id: int,
  1539. db: AsyncSession = Depends(get_db),
  1540. current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_DELETE_ALL),
  1541. ):
  1542. """Cancel all pending items in a batch and mark batch as cancelled."""
  1543. result = await db.execute(select(PrintBatch).where(PrintBatch.id == batch_id))
  1544. batch = result.scalar_one_or_none()
  1545. if not batch:
  1546. raise HTTPException(404, "Batch not found")
  1547. # Cancel all pending queue items in this batch
  1548. result = await db.execute(
  1549. select(PrintQueueItem).where(and_(PrintQueueItem.batch_id == batch_id, PrintQueueItem.status == "pending"))
  1550. )
  1551. pending_items = result.scalars().all()
  1552. cancelled_count = 0
  1553. cancelled_ids: list[int] = []
  1554. for item in pending_items:
  1555. item.status = "cancelled"
  1556. await release_budget_reservation(
  1557. db,
  1558. source_type="print_queue",
  1559. source_id=item.id,
  1560. status="released",
  1561. )
  1562. cancelled_ids.append(item.id)
  1563. cancelled_count += 1
  1564. batch.status = "cancelled"
  1565. await db.commit()
  1566. # Same as the single-item path: a dispatch already preheating for one of
  1567. # these cannot see the status change on its own (#2727).
  1568. from backend.app.services.print_scheduler import scheduler as _scheduler
  1569. for _cancelled_id in cancelled_ids:
  1570. _scheduler.notify_dispatch_cancelled(_cancelled_id)
  1571. return {"message": f"Batch cancelled, {cancelled_count} pending items cancelled"}
  1572. async def _build_batch_response(
  1573. db: AsyncSession, batch: PrintBatch, *, usernames: dict[int, str] | None = None
  1574. ) -> PrintBatchResponse:
  1575. """Build a batch response with per-plate progress derived from queue items.
  1576. ``usernames`` lets the list endpoint resolve every creator in one query
  1577. instead of one per batch.
  1578. """
  1579. progress = await load_progress(db, batch)
  1580. created_by_username = None
  1581. if batch.created_by_id:
  1582. if usernames is not None:
  1583. created_by_username = usernames.get(batch.created_by_id)
  1584. else:
  1585. result = await db.execute(select(User).where(User.id == batch.created_by_id))
  1586. user = result.scalar_one_or_none()
  1587. if user:
  1588. created_by_username = user.username
  1589. return PrintBatchResponse(
  1590. id=batch.id,
  1591. name=batch.name,
  1592. archive_id=batch.archive_id,
  1593. library_file_id=batch.library_file_id,
  1594. quantity=batch.quantity,
  1595. status=batch.status,
  1596. created_at=batch.created_at,
  1597. completed_at=batch.completed_at,
  1598. created_by_id=batch.created_by_id,
  1599. created_by_username=created_by_username,
  1600. project_id=batch.project_id,
  1601. due_date=batch.due_date,
  1602. notes=batch.notes,
  1603. pending_count=progress.pending,
  1604. printing_count=progress.printing,
  1605. completed_count=progress.completed,
  1606. failed_count=progress.failed,
  1607. cancelled_count=progress.cancelled,
  1608. skipped_count=progress.skipped,
  1609. has_targets=progress.has_targets,
  1610. target_count=progress.target,
  1611. remaining_count=progress.remaining,
  1612. dispatchable_count=progress.dispatchable_remaining,
  1613. actual_cost=progress.actual_cost,
  1614. estimated_remaining_cost=progress.estimated_remaining_cost,
  1615. filament_used_grams=progress.filament_used_grams,
  1616. print_time_seconds=progress.print_time_seconds,
  1617. plates=[
  1618. PrintBatchPlateProgress(
  1619. plate_id=plate.plate_id,
  1620. plate_name=plate.plate_name,
  1621. quantity_target=plate.quantity_target,
  1622. dispatched=plate.dispatched,
  1623. remaining=plate.remaining,
  1624. pending_count=plate.pending,
  1625. printing_count=plate.printing,
  1626. completed_count=plate.completed,
  1627. failed_count=plate.failed,
  1628. cancelled_count=plate.cancelled,
  1629. skipped_count=plate.skipped,
  1630. actual_cost=plate.actual_cost,
  1631. estimated_remaining_cost=plate.estimated_remaining_cost,
  1632. filament_used_grams=plate.filament_used_grams,
  1633. print_time_seconds=plate.print_time_seconds,
  1634. can_dispatch=plate.can_dispatch,
  1635. )
  1636. for plate in progress.plates
  1637. ],
  1638. )
  1639. @router.get("/{item_id}", response_model=PrintQueueItemResponse)
  1640. async def get_queue_item(
  1641. item_id: int,
  1642. db: AsyncSession = Depends(get_db),
  1643. auth_result: tuple[User | None, bool] = Depends(
  1644. require_ownership_permission(
  1645. Permission.QUEUE_READ_ALL,
  1646. Permission.QUEUE_READ_OWN,
  1647. )
  1648. ),
  1649. ):
  1650. """Get a specific queue item."""
  1651. current_user, can_read_all = auth_result
  1652. result = await db.execute(
  1653. select(PrintQueueItem)
  1654. .options(
  1655. selectinload(PrintQueueItem.archive),
  1656. selectinload(PrintQueueItem.printer),
  1657. selectinload(PrintQueueItem.library_file),
  1658. selectinload(PrintQueueItem.created_by),
  1659. selectinload(PrintQueueItem.batch),
  1660. # Cross-model candidates (#671) and their files, for the card label.
  1661. selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
  1662. )
  1663. .where(PrintQueueItem.id == item_id)
  1664. )
  1665. item = result.scalar_one_or_none()
  1666. if not item:
  1667. raise HTTPException(404, "Queue item not found")
  1668. if (
  1669. current_user is not None
  1670. and not can_read_all
  1671. and (item.created_by_id is None or item.created_by_id != current_user.id)
  1672. ):
  1673. raise HTTPException(404, "Queue item not found")
  1674. return _enrich_response(item)
  1675. @router.patch("/{item_id}", response_model=PrintQueueItemResponse)
  1676. async def update_queue_item(
  1677. item_id: int,
  1678. data: PrintQueueItemUpdate,
  1679. db: AsyncSession = Depends(get_db),
  1680. auth_result: tuple[User | None, bool] = Depends(
  1681. require_ownership_permission(
  1682. Permission.QUEUE_UPDATE_ALL,
  1683. Permission.QUEUE_UPDATE_OWN,
  1684. )
  1685. ),
  1686. ):
  1687. """Update a queue item."""
  1688. user, can_modify_all = auth_result
  1689. result = await db.execute(
  1690. select(PrintQueueItem)
  1691. # Needed by the cross-model guard below, and by the response builder —
  1692. # without it _variant_summaries falls back to [] and a PATCH would strip
  1693. # the alternatives out of the payload it echoes back.
  1694. .options(selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file))
  1695. .where(PrintQueueItem.id == item_id)
  1696. )
  1697. item = result.scalar_one_or_none()
  1698. if not item:
  1699. raise HTTPException(404, "Queue item not found")
  1700. # Ownership check
  1701. if not can_modify_all:
  1702. if item.created_by_id != user.id:
  1703. raise HTTPException(403, "You can only update your own queue items")
  1704. if item.status != "pending":
  1705. raise HTTPException(400, "Can only update pending items")
  1706. # Dispatch claim (#2615): the row is pending but a scheduler worker has
  1707. # already claimed it and is uploading to its printer. Editing now (e.g.
  1708. # reassigning printer_id) would split the queue row from the in-flight
  1709. # archive/expected-print/physical command. Reject until dispatch finishes;
  1710. # to move it, cancel first (the coordinated escape) and re-queue.
  1711. if item.dispatching_at is not None:
  1712. raise HTTPException(409, "Item is being dispatched — cancel it first to make changes")
  1713. update_data = data.model_dump(exclude_unset=True)
  1714. # Normalize target_model if being updated (see add_to_queue for why the
  1715. # code map has to run first).
  1716. if "target_model" in update_data and update_data["target_model"]:
  1717. update_data["target_model"] = normalize_model_name(update_data["target_model"])
  1718. # A cross-model item (#671) owns its own printer decision: each candidate
  1719. # carries its model, and the resolver folds the winner onto the row at
  1720. # dispatch. Assigning a printer here would leave a row with variants *and* a
  1721. # printer_id, and the fixed-printer branch of the scheduler wins that race —
  1722. # so it would dispatch a row whose library_file_id is still null and die in
  1723. # the upload. Narrowing target_model is refused for the same reason: it
  1724. # would silently discard every alternative the user queued.
  1725. #
  1726. # Compared against the current value rather than merely present, because the
  1727. # edit dialog re-sends target_model unchanged on every save.
  1728. if item.variants:
  1729. for field in ("printer_id", "target_model"):
  1730. if field in update_data and update_data[field] != getattr(item, field):
  1731. raise HTTPException(
  1732. 400,
  1733. "This job has printer alternatives — remove them before assigning a printer or model",
  1734. )
  1735. # Cannot specify both printer_id and target_model
  1736. new_printer_id = update_data.get("printer_id", item.printer_id)
  1737. new_target_model = update_data.get("target_model", item.target_model)
  1738. if new_printer_id and new_target_model:
  1739. raise HTTPException(400, "Cannot specify both printer_id and target_model")
  1740. # Validate new printer_id if being changed (and not None)
  1741. if "printer_id" in update_data and update_data["printer_id"] is not None:
  1742. result = await db.execute(select(Printer).where(Printer.id == update_data["printer_id"]))
  1743. if not result.scalar_one_or_none():
  1744. raise HTTPException(400, "Printer not found")
  1745. # Validate target_model has active printers
  1746. if "target_model" in update_data and update_data["target_model"]:
  1747. result = await db.execute(
  1748. select(Printer).where(Printer.model == update_data["target_model"]).where(Printer.is_active == True) # noqa: E712
  1749. )
  1750. if not result.scalars().first():
  1751. raise HTTPException(400, f"No active printers for model: {update_data['target_model']}")
  1752. # Cross-model safety gate (#2578) — same check as the create route, so
  1753. # a mismatched target can't be introduced by editing either.
  1754. sliced_for = None
  1755. if item.archive_id:
  1756. result = await db.execute(select(PrintArchive.sliced_for_model).where(PrintArchive.id == item.archive_id))
  1757. sliced_for = result.scalar_one_or_none()
  1758. elif item.library_file_id:
  1759. result = await db.execute(select(LibraryFile).where(LibraryFile.id == item.library_file_id))
  1760. lib = result.scalar_one_or_none()
  1761. if lib and lib.file_metadata:
  1762. sliced_for = lib.file_metadata.get("sliced_for_model")
  1763. if not is_gcode_compatible(sliced_for, update_data["target_model"]):
  1764. raise HTTPException(
  1765. 400,
  1766. f"File was sliced for {sliced_for} and cannot be dispatched to {update_data['target_model']} printers",
  1767. )
  1768. # Serialize ams_mapping to JSON for TEXT column storage
  1769. if "ams_mapping" in update_data:
  1770. update_data["ams_mapping"] = json.dumps(update_data["ams_mapping"]) if update_data["ams_mapping"] else None
  1771. # Serialize filament_overrides to JSON for TEXT column storage, keeping only
  1772. # the slots this item's plate actually prints (#2551 — same shared-override
  1773. # list the create path narrows).
  1774. if "filament_overrides" in update_data:
  1775. overrides = update_data["filament_overrides"]
  1776. if overrides:
  1777. overrides = overrides_for_plate(
  1778. overrides,
  1779. await _resolve_source_path(db, item),
  1780. update_data.get("plate_id", item.plate_id),
  1781. )
  1782. update_data["filament_overrides"] = json.dumps(overrides) if overrides else None
  1783. # Serialize H2C rack-swap nozzle pick (#1780) to JSON for TEXT column
  1784. # storage; same Text-as-opaque-blob convention as ams_mapping above.
  1785. if "nozzle_mapping" in update_data:
  1786. update_data["nozzle_mapping"] = (
  1787. json.dumps(update_data["nozzle_mapping"]) if update_data["nozzle_mapping"] else None
  1788. )
  1789. # Same Text-as-JSON convention for the rack-position pick (#1784). An empty
  1790. # object clears it, which is how the UI says "assign these for me again".
  1791. if "nozzle_rack_choice" in update_data:
  1792. update_data["nozzle_rack_choice"] = (
  1793. json.dumps(update_data["nozzle_rack_choice"]) if update_data["nozzle_rack_choice"] else None
  1794. )
  1795. trusted_estimated_cost = await _trusted_item_estimated_cost(
  1796. db,
  1797. item,
  1798. printer_id=update_data.get("printer_id", item.printer_id),
  1799. plate_id=update_data.get("plate_id", item.plate_id),
  1800. ams_mapping=update_data.get("ams_mapping", item.ams_mapping),
  1801. )
  1802. update_data["estimated_cost"] = trusted_estimated_cost
  1803. await validate_print_budget(
  1804. db,
  1805. cost_center_id=update_data.get("cost_center_id", item.cost_center_id),
  1806. estimated_cost=trusted_estimated_cost,
  1807. current_user=user,
  1808. exclude_queue_item_id=item.id,
  1809. )
  1810. # Re-check the dispatch claim right before mutating (#2615). Several awaited
  1811. # validations ran since the guard above, and a scheduler worker may have
  1812. # claimed the row in that gap. A fresh read (item isn't dirty yet, so no
  1813. # autoflush races the check) narrows the window to effectively nothing.
  1814. claimed = (
  1815. await db.execute(select(PrintQueueItem.dispatching_at).where(PrintQueueItem.id == item_id))
  1816. ).scalar_one_or_none()
  1817. if claimed is not None:
  1818. raise HTTPException(409, "Item is being dispatched — cancel it first to make changes")
  1819. for field, value in update_data.items():
  1820. setattr(item, field, value)
  1821. await db.commit()
  1822. await db.refresh(item, ["archive", "printer", "library_file", "created_by", "batch"])
  1823. logger.info("Updated queue item %s", item_id)
  1824. return _enrich_response(item)
  1825. @router.delete("/{item_id}")
  1826. async def delete_queue_item(
  1827. item_id: int,
  1828. db: AsyncSession = Depends(get_db),
  1829. auth_result: tuple[User | None, bool] = Depends(
  1830. require_ownership_permission(
  1831. Permission.QUEUE_DELETE_ALL,
  1832. Permission.QUEUE_DELETE_OWN,
  1833. )
  1834. ),
  1835. ):
  1836. """Remove an item from the queue.
  1837. An order's last surviving run for a plate is cancelled instead of deleted
  1838. (#2960). The row is what a later dispatch clones, so removing it would
  1839. leave the order reporting work outstanding that nothing could ever
  1840. produce. A completed run is exempt: it is the record of something that was
  1841. actually made, and rewriting it as cancelled would falsify the order's
  1842. progress.
  1843. """
  1844. user, can_modify_all = auth_result
  1845. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
  1846. item = result.scalar_one_or_none()
  1847. if not item:
  1848. raise HTTPException(404, "Queue item not found")
  1849. # Ownership check
  1850. if not can_modify_all:
  1851. if item.created_by_id != user.id:
  1852. raise HTTPException(403, "You can only delete your own queue items")
  1853. if item.status == "printing":
  1854. raise HTTPException(400, "Cannot delete item that is currently printing")
  1855. keep_as_cancelled = item.status != "completed" and await _is_orders_last_source(db, item)
  1856. await release_budget_reservation(
  1857. db,
  1858. source_type="print_queue",
  1859. source_id=item.id,
  1860. status="released",
  1861. )
  1862. if keep_as_cancelled:
  1863. item.status = "cancelled"
  1864. await db.flush()
  1865. # The order may have been sitting on "completed" if this run's target
  1866. # was met by it; cancelling reopens it.
  1867. await refresh_batch_status_for_item(db, item.id)
  1868. else:
  1869. await db.delete(item)
  1870. await db.commit()
  1871. # Stop an in-flight preheat for this item: the dispatch coroutine is
  1872. # parked in a sleep and cannot see the status we just wrote (#2727).
  1873. from backend.app.services.print_scheduler import scheduler as _scheduler
  1874. _scheduler.notify_dispatch_cancelled(item_id)
  1875. if keep_as_cancelled:
  1876. logger.info("Kept queue item %s as cancelled — last source for its batch order plate", item_id)
  1877. return {
  1878. "message": "Item cancelled rather than deleted: it is the only run the order can re-queue this plate from",
  1879. "deleted": False,
  1880. }
  1881. logger.info("Deleted queue item %s", item_id)
  1882. return {"message": "Queue item deleted", "deleted": True}
  1883. @router.post("/reorder")
  1884. async def reorder_queue(
  1885. data: PrintQueueReorder,
  1886. db: AsyncSession = Depends(get_db),
  1887. _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_ALL),
  1888. ):
  1889. """Bulk update positions for queue items."""
  1890. for reorder_item in data.items:
  1891. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == reorder_item.id))
  1892. item = result.scalar_one_or_none()
  1893. if item and item.status == "pending":
  1894. item.position = reorder_item.position
  1895. await db.commit()
  1896. logger.info("Reordered %s queue items", len(data.items))
  1897. return {"message": f"Reordered {len(data.items)} items"}
  1898. @router.post("/printer/{printer_id}/resume")
  1899. async def resume_queue_after_failure(
  1900. printer_id: int,
  1901. db: AsyncSession = Depends(get_db),
  1902. _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_ALL),
  1903. ):
  1904. """Clear the previous-success gate for a printer and restore skipped items.
  1905. Single atomic op (#1818):
  1906. * Sets ``gate_acknowledged=True`` on every ``failed`` / ``aborted`` queue
  1907. item for this printer that's still in the scheduler's lookback window,
  1908. so the next ``_check_previous_success`` call ignores them.
  1909. * Restores ``skipped`` items whose ``error_message`` matches the
  1910. scheduler's exact "Previous print failed or was aborted" gate string
  1911. back to ``pending`` (clears ``error_message`` + ``completed_at``).
  1912. Returns counts so the UI can render a precise toast. No-op endpoint
  1913. (zero counts) when called against a printer with no gate to clear.
  1914. """
  1915. result = await db.execute(select(Printer).where(Printer.id == printer_id))
  1916. printer = result.scalar_one_or_none()
  1917. if not printer:
  1918. raise HTTPException(404, "Printer not found")
  1919. ack_result = await db.execute(
  1920. select(PrintQueueItem)
  1921. .where(PrintQueueItem.printer_id == printer_id)
  1922. .where(PrintQueueItem.status.in_(["failed", "aborted"]))
  1923. .where(PrintQueueItem.gate_acknowledged == False) # noqa: E712
  1924. )
  1925. to_ack = ack_result.scalars().all()
  1926. for failed_item in to_ack:
  1927. failed_item.gate_acknowledged = True
  1928. restore_result = await db.execute(
  1929. select(PrintQueueItem)
  1930. .where(PrintQueueItem.printer_id == printer_id)
  1931. .where(PrintQueueItem.status == "skipped")
  1932. .where(PrintQueueItem.error_message == "Previous print failed or was aborted")
  1933. )
  1934. to_restore = restore_result.scalars().all()
  1935. for skipped_item in to_restore:
  1936. skipped_item.status = "pending"
  1937. skipped_item.error_message = None
  1938. skipped_item.completed_at = None
  1939. await db.commit()
  1940. logger.info(
  1941. "Resume after failure on printer %s: acknowledged %d failure(s), restored %d skipped item(s)",
  1942. printer_id,
  1943. len(to_ack),
  1944. len(to_restore),
  1945. )
  1946. return {"acknowledged": len(to_ack), "restored": len(to_restore)}
  1947. @router.post("/{item_id}/cancel")
  1948. async def cancel_queue_item(
  1949. item_id: int,
  1950. db: AsyncSession = Depends(get_db),
  1951. auth_result: tuple[User | None, bool] = Depends(
  1952. require_ownership_permission(
  1953. Permission.QUEUE_UPDATE_ALL,
  1954. Permission.QUEUE_UPDATE_OWN,
  1955. )
  1956. ),
  1957. ):
  1958. """Cancel a pending queue item."""
  1959. user, can_modify_all = auth_result
  1960. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
  1961. item = result.scalar_one_or_none()
  1962. if not item:
  1963. raise HTTPException(404, "Queue item not found")
  1964. # Ownership check
  1965. if not can_modify_all:
  1966. if item.created_by_id != user.id:
  1967. raise HTTPException(403, "You can only cancel your own queue items")
  1968. if item.status not in ("pending",):
  1969. raise HTTPException(400, f"Cannot cancel item with status '{item.status}'")
  1970. item.status = "cancelled"
  1971. item.completed_at = datetime.now(timezone.utc)
  1972. await release_budget_reservation(
  1973. db,
  1974. source_type="print_queue",
  1975. source_id=item.id,
  1976. status="released",
  1977. )
  1978. await db.commit()
  1979. # Stop an in-flight preheat for this item: the dispatch coroutine is
  1980. # parked in a sleep and cannot see the status we just wrote (#2727).
  1981. from backend.app.services.print_scheduler import scheduler as _scheduler
  1982. _scheduler.notify_dispatch_cancelled(item_id)
  1983. logger.info("Cancelled queue item %s", item_id)
  1984. return {"message": "Queue item cancelled"}
  1985. @router.post("/{item_id}/stop")
  1986. async def stop_queue_item(
  1987. item_id: int,
  1988. db: AsyncSession = Depends(get_db),
  1989. auth_result: tuple[User | None, bool] = Depends(
  1990. require_ownership_permission(
  1991. Permission.QUEUE_UPDATE_ALL,
  1992. Permission.QUEUE_UPDATE_OWN,
  1993. )
  1994. ),
  1995. ):
  1996. """Stop an actively printing queue item.
  1997. Ownership-scoped (#1625-followup): callers with QUEUE_UPDATE_OWN can stop
  1998. their own items; callers with QUEUE_UPDATE_ALL can stop any item. Mirrors
  1999. the /cancel shape. Pre-fix this required QUEUE_UPDATE_ALL — Operators
  2000. holding only _OWN saw the Stop button in the queue UI but got 403 on click.
  2001. """
  2002. from backend.app.services.printer_manager import printer_manager
  2003. user, can_modify_all = auth_result
  2004. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
  2005. item = result.scalar_one_or_none()
  2006. if not item:
  2007. raise HTTPException(404, "Queue item not found")
  2008. # Ownership check — mirrors /cancel. Ownerless items (created_by_id IS NULL)
  2009. # require _ALL: stop is destructive and an _OWN holder can't claim "they
  2010. # own it" the way /start does (#1670).
  2011. if not can_modify_all and user is not None:
  2012. if item.created_by_id is None or item.created_by_id != user.id:
  2013. raise HTTPException(403, "You can only stop your own queue items")
  2014. if item.status != "printing":
  2015. raise HTTPException(400, f"Can only stop items that are printing, current status: '{item.status}'")
  2016. # Capture values we need for background task
  2017. printer_id = item.printer_id
  2018. auto_off_after = item.auto_off_after
  2019. # Try to send stop command to printer
  2020. stop_sent = False
  2021. try:
  2022. stop_sent = printer_manager.stop_print(printer_id)
  2023. if not stop_sent:
  2024. logger.warning("stop_print returned False for printer %s - printer may not be connected", printer_id)
  2025. except Exception as e:
  2026. logger.error("Error sending stop command for queue item %s: %s", item_id, e)
  2027. # Mark this printer as user-stopped BEFORE the first await so that if the
  2028. # MQTT on_print_complete callback fires during the db.commit() yield the flag
  2029. # is already set and the "failed" status will be correctly overridden to
  2030. # "cancelled" (preventing a spurious "print failed" notification).
  2031. try:
  2032. from backend.app.main import mark_printer_stopped_by_user
  2033. mark_printer_stopped_by_user(printer_id)
  2034. except Exception as _mark_err:
  2035. logger.warning("Failed to mark printer %s as user-stopped: %s", printer_id, _mark_err)
  2036. # Update queue item status regardless - if printer is off, print is already stopped
  2037. item.status = "cancelled"
  2038. item.completed_at = datetime.now(timezone.utc)
  2039. item.error_message = "Stopped by user" if stop_sent else "Stopped by user (printer was offline)"
  2040. # Reconcile the linked archive when the printer is offline (#2603). When the
  2041. # stop command reaches the printer it later reports the stop over MQTT and
  2042. # on_print_complete flips the archive to cancelled/failed. When the printer is
  2043. # offline no such event ever arrives, so the archive would stay "printing"
  2044. # forever (queue row cancelled, archive still printing — the reporter's
  2045. # archive 436). Close it out here, mirroring what the MQTT path would have
  2046. # done. Only touch a still-"printing" archive so we never overwrite a real
  2047. # completion that raced in.
  2048. if not stop_sent and item.archive_id:
  2049. archive = await db.get(PrintArchive, item.archive_id)
  2050. if archive and archive.status == "printing":
  2051. archive.status = "cancelled"
  2052. archive.completed_at = datetime.now(timezone.utc)
  2053. archive.failure_reason = "Stopped by user (printer was offline)"
  2054. await db.commit()
  2055. logger.info("Stopped printing queue item %s (stop command sent: %s)", item_id, stop_sent)
  2056. # Schedule power-off if the queue item opted in. Delegates to the smart-plug
  2057. # manager so the off honours each plug's configured strategy (time delay or
  2058. # temperature threshold), is cancelled if the printer starts printing again,
  2059. # and never cuts power on a loaded print (#1890). Previously an inline block
  2060. # hardcoded a 50°C / 600s cooldown wait and powered off on the timeout
  2061. # regardless of print state.
  2062. if auto_off_after:
  2063. from backend.app.services.smart_plug_manager import smart_plug_manager
  2064. try:
  2065. await smart_plug_manager.schedule_off_after_queue_job(printer_id, db)
  2066. except Exception as e:
  2067. logger.warning("Auto-off: Failed to schedule power-off for printer %s: %s", printer_id, e)
  2068. return {"message": "Print stopped" if stop_sent else "Queue item cancelled (printer was offline)"}
  2069. @router.post("/{item_id}/start")
  2070. async def start_queue_item(
  2071. item_id: int,
  2072. skip_filament_check: bool = Query(default=False),
  2073. db: AsyncSession = Depends(get_db),
  2074. auth_result: tuple[User | None, bool] = Depends(
  2075. require_ownership_permission(
  2076. Permission.QUEUE_UPDATE_ALL,
  2077. Permission.QUEUE_UPDATE_OWN,
  2078. )
  2079. ),
  2080. ):
  2081. """Manually start a staged (manual_start) queue item.
  2082. Ownership-scoped (#1625-followup): callers with QUEUE_UPDATE_OWN can
  2083. start their own items + claim ownership of NULL-owner items (VP-uploaded
  2084. items arrive unattributed per #1670). Callers with QUEUE_UPDATE_ALL can
  2085. start any item. Pre-fix this required QUEUE_UPDATE_OWN with no ownership
  2086. check, so _OWN holders could start anyone's queue items via direct API.
  2087. Clears the manual_start flag so the scheduler picks it up. When
  2088. ``skip_filament_check`` is false (the default) the live filament
  2089. deficit (#1496) is checked first — if the assigned spool can't satisfy
  2090. a slot's required grams, the route returns ``409`` with the deficit
  2091. payload so the caller can show a confirm dialog and retry with
  2092. ``skip_filament_check=true``.
  2093. """
  2094. user, can_modify_all = auth_result
  2095. result = await db.execute(
  2096. select(PrintQueueItem)
  2097. .options(
  2098. selectinload(PrintQueueItem.archive),
  2099. selectinload(PrintQueueItem.printer),
  2100. selectinload(PrintQueueItem.library_file),
  2101. selectinload(PrintQueueItem.batch),
  2102. selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
  2103. )
  2104. .where(PrintQueueItem.id == item_id)
  2105. )
  2106. item = result.scalar_one_or_none()
  2107. if not item:
  2108. raise HTTPException(404, "Queue item not found")
  2109. # Ownership check — softer than /cancel because /start is the entry point
  2110. # for #1670's VP-import flow: an unowned item is claimable by the first
  2111. # _OWN holder who clicks ▶, and the route below credits them as owner.
  2112. # An item with a DIFFERENT owner → 403.
  2113. if not can_modify_all and user is not None:
  2114. if item.created_by_id is not None and item.created_by_id != user.id:
  2115. raise HTTPException(403, "You can only start your own queue items")
  2116. if item.status != "pending":
  2117. raise HTTPException(400, f"Can only start pending items, current status: '{item.status}'")
  2118. item.estimated_cost = await _trusted_item_estimated_cost(
  2119. db,
  2120. item,
  2121. printer_id=item.printer_id,
  2122. plate_id=item.plate_id,
  2123. ams_mapping=item.ams_mapping,
  2124. )
  2125. await validate_print_budget(
  2126. db,
  2127. cost_center_id=item.cost_center_id,
  2128. estimated_cost=item.estimated_cost,
  2129. current_user=user,
  2130. exclude_queue_item_id=item.id,
  2131. )
  2132. # Live deficit check — re-evaluated against current spool state, so a
  2133. # spool swap between scheduler flagging and the user clicking ▶ clears
  2134. # the block automatically.
  2135. if not skip_filament_check:
  2136. deficit = await compute_deficit_for_queue_item(db, item)
  2137. if deficit:
  2138. raise HTTPException(
  2139. status_code=409,
  2140. detail={
  2141. "code": "insufficient_filament",
  2142. "deficit": [d.to_dict() for d in deficit],
  2143. },
  2144. )
  2145. # Print Anyway / no deficit: clear the flags and let the scheduler dispatch.
  2146. item.manual_start = False
  2147. item.filament_short = False
  2148. # Persist the user's "Print Anyway" decision so the scheduler does not
  2149. # immediately re-flag this item on the next tick (#1698-followup). The
  2150. # pre-fix behaviour bounced between "user said anyway" and
  2151. # "scheduler re-blocked on same deficit" forever.
  2152. if skip_filament_check:
  2153. item.skip_filament_check = True
  2154. # Credit the clicker as the item's owner when no prior owner is set —
  2155. # VP-uploaded queue items arrive over FTP unattributed, so without this
  2156. # the print log's User column stays blank even when auth is on
  2157. # (#1670). An item that already has a creator (UI-added queue items)
  2158. # keeps that attribution; the dispatcher is not promoted over the
  2159. # original uploader.
  2160. if user is not None and item.created_by_id is None:
  2161. item.created_by_id = user.id
  2162. await db.commit()
  2163. await db.refresh(item, ["archive", "printer", "library_file", "created_by", "batch"])
  2164. logger.info(
  2165. "Manually started queue item %s (cleared manual_start; skip_filament_check=%s)",
  2166. item_id,
  2167. skip_filament_check,
  2168. )
  2169. return _enrich_response(item)