feat(printer-models): Job lifecycle FSM with explicit state machine#49
Conversation
…nd terminal-event signalling Implements JobState (StrEnum), Job (dataclass), JobStateMachine, and InvalidStateTransitionError covering all valid transitions per the Brother raster spec (no mid-print cancel/pause). Terminal states signal asyncio.Event _done_event for the upcoming wait_for_job() in PR B (PrintQueue). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…terminal-state coverage - image_payload: field(repr=False) — avoids spamming logs with raw bytes - _done_event: field(init=False, repr=False) — makes it truly internal; Job(..., _done_event=x) now raises TypeError instead of silently bypassing intended construction - Add three tests: FAILED and CANCELLED set _done_event + finished_at, and both terminal states absorb all outgoing transitions Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request establishes the core state management logic for print jobs in isolation. By implementing a strict FSM, it ensures that job transitions adhere to hardware constraints, specifically preventing invalid mid-print operations. This foundational work provides the necessary structures for the upcoming print queue worker implementation. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
There was a problem hiding this comment.
Code Review
This pull request introduces a job lifecycle finite-state machine for the print queue, including a JobState enum, a Job dataclass, and a JobStateMachine to manage transitions and side effects. It also adds comprehensive unit tests for the state machine logic. Feedback focuses on replacing naive datetimes with timezone-aware UTC timestamps for consistency and addressing a type safety violation regarding the use of Any in the job options dictionary.
|
|
||
| import asyncio | ||
| from dataclasses import dataclass, field | ||
| from datetime import datetime |
There was a problem hiding this comment.
| id: str | ||
| printer_id: str | ||
| state: JobState = JobState.QUEUED | ||
| submitted_at: datetime = field(default_factory=datetime.now) |
There was a problem hiding this comment.
Using naive datetime.now() can lead to issues when comparing timestamps across different environments or during daylight saving changes. Use datetime.now(UTC) for consistency.
| submitted_at: datetime = field(default_factory=datetime.now) | |
| submitted_at: datetime = field(default_factory=lambda: datetime.now(UTC)) |
| finished_at: datetime | None = None | ||
| image_payload: bytes | None = field(default=None, repr=False) | ||
| tape_mm: int | None = None | ||
| options: dict[str, Any] = field(default_factory=dict) |
There was a problem hiding this comment.
This introduces Any, which is flagged by the repository's strict type safety policy (Priority 7). Consider using a more specific type for printer options, such as dict[str, str | int | bool | float | None], or define a TypeAlias if the schema is known.
References
- Flag new Any introductions. mypy --strict on app/. (link)
| job.state = new_state | ||
|
|
||
| if new_state == JobState.PRINTING and job.started_at is None: | ||
| job.started_at = datetime.now() |
| job.started_at = datetime.now() | ||
|
|
||
| if new_state in _TERMINAL_STATES: | ||
| job.finished_at = datetime.now() |
There was a problem hiding this comment.
Pull request overview
This PR introduces an explicit finite-state machine for per-print-job lifecycle tracking in the backend, intended to be consumed by the upcoming PrintQueue worker (PR B) to enforce Brother-spec-compliant transitions and centralize timestamp/event side effects.
Changes:
- Added
JobState,Job,InvalidStateTransitionError, andJobStateMachine.transition()to model and enforce legal job state transitions. - Implemented terminal-state side effects (finished timestamp + done-event signaling) and printing-start timestamping.
- Added a unit test suite covering valid/invalid transitions and done-event behavior for terminal vs non-terminal transitions.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
backend/app/services/job_lifecycle.py |
Adds the Job model and FSM transition gatekeeper with timestamps and terminal completion signaling. |
backend/tests/unit/services/test_job_lifecycle.py |
Adds unit tests for transition validity and _done_event signaling invariants. |
| finished_at: datetime | None = None | ||
| image_payload: bytes | None = field(default=None, repr=False) | ||
| tape_mm: int | None = None | ||
| options: dict[str, Any] = field(default_factory=dict) |
| # asyncio.Event is allowed on a dataclass field because asyncio fields on | ||
| # Jobs are only awaited from within an event loop; constructing them at | ||
| # import time is OK since asyncio.Event does not require a running loop. |
…t comment - Import UTC from datetime; replace all datetime.now() calls with datetime.now(UTC) (submitted_at default_factory, started_at and finished_at in transition()) so timestamps are timezone-aware and safe across DST boundaries and SQLModel persistence (Phase 5). - Rewrite the _done_event comment: it was wrong about "import time" — default_factory runs at instance creation, and since Python 3.10 asyncio.Event does not pin to a loop at construction. - Add rationale comment on options: dict[str, Any] so future reviewers don't re-flag the intentional Any. - Add test_timestamps_are_utc_aware() verifying tzinfo is UTC on all three timestamp fields (14th test, 64 total). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
## 0.3.0 (2026-05-12) * feat(config): pydantic-settings module with env-driven runtime configuration (#45) ([878e9e0](878e9e0)), closes [#45](#45) * feat(integrations): AppLookupService aggregator — Phase 3 complete (#53) ([222bef4](222bef4)), closes [#53](#53) * feat(integrations): Grocy + Spoolman lookup clients with shared NotFoundError base (#52) ([b1c9c3c](b1c9c3c)), closes [#52](#52) * feat(integrations): LabelData schema + Snipe-IT lookup client (#51) ([3bc180f](3bc180f)), closes [#51](#51) * feat(label-renderer): Template schema + Pillow/qrcode renderer for 1-bit label bitmaps (#54) ([fb77028](fb77028)), closes [#54](#54) * feat(printer-models): Brother PT-Series TapeRegistry with TZe and heat-shrink specs (#47) ([7526019](7526019)), closes [#47](#47) * feat(printer-models): Job lifecycle FSM with explicit state machine (#49) ([1a8c40e](1a8c40e)), closes [#49](#49) * feat(printer-models): PrinterModel Protocol + ModelRegistry for plugin discovery (#48) ([2ae0e09](2ae0e09)), closes [#48](#48) * feat(printer-models): PrintQueue worker with pause/resume/cancel/retry (#50) ([dfdf6fe](dfdf6fe)), closes [#50](#50) [skip ci]
Summary
PR A of split-Task 2.8. Adds the per-print-job state machine in isolation. The PrintQueue worker (PR B) imports
Job,JobState,JobStateMachine, andInvalidStateTransitionErrorand drives transitions from its async loop.What's in this PR
backend/app/services/job_lifecycle.pyJobState(StrEnum):QUEUED,PAUSED,PRINTING,COMPLETED,FAILED,CANCELLED._VALID_TRANSITIONS: dict[JobState, frozenset[JobState]]enumerates every legal edge. PRINTING is one-way toCOMPLETED/FAILEDper the Brother Raster Spec — the printer ignores commands once the raster stream starts, so the FSM mirrors that constraint._TERMINAL_STATES: frozenset[JobState]extracted to keep the side-effect logic DRY.@dataclass Jobwith 12 fields.image_payloadisrepr=False(a real raster image is ~30 kB; default repr would shred log lines)._done_eventisinit=False, repr=False— internal plumbing, not for caller construction.InvalidStateTransitionError(Exception)— message names both states for log self-containment.JobStateMachine.transition(job, new_state)— stateless@staticmethodgatekeeper. Setsstarted_atonPRINTING, setsfinished_atand_done_eventon any terminal state.backend/tests/unit/services/test_job_lifecycle.py— 13 tests_done_event.set()on FAILED, this locks in the invariantWhat's NOT in this PR
PrintQueueclass, noPrinterWorkerState, noasyncio.Queue, no worker code — those land in PR B on top of this branch's symbols.Test plan
pytest -q— 63/63 (50 prior + 13 new).ruff format --check .clean.ruff check .clean.mypy app/(strict) clean.Job(id='x', printer_id='y')repr does not leakimage_payloador_done_event.Job(..., _done_event=other)raisesTypeError(init guard works).Review history (subagent-driven)
b0e4dab— initial commit.InvalidStateTransition→InvalidStateTransitionErrorto satisfy ruff N818.image_payload, init-leak of_done_event, missing terminal-event coverage.a44708b— repr/init suppression + 3 terminal-coverage tests.Linked plan
docs/superpowers/plans/2026-05-11-label-printer-hub.mdTask 2.8 (Phase 2 — split A of 2). PR B (PrintQueue worker) follows on top.