kawa
356c2ee690
- imptune/api/printers.py: POST /printers (all form fields, checkbox->bool,
FK resolution), DELETE /printers/{id}, grouped list renderer with LEFT JOIN
- imptune/api/clients.py: POST /clients with duplicate-name handling
- imptune/api/pages.py: GET /printers and GET /clients page routes
- imptune/main.py: register printers + clients routers; close db on shutdown
- imptune/db/database.py: close existing connection before re-init (test isolation)
- templates: printers.html, clients.html, printer_form.html (Alpine.js port
auto-derivation), printer_list.html (grouped by client), client_list.html
- tests/conftest.py: close test-thread db connection in fixture teardown
- tests/test_printer_crud.py: updated to use list(select().where()) for DB
queries (avoids Peewee thread-local cursor caching across test boundaries)
Auto-fix [Rule 1 - Bug]: DB test isolation — Peewee thread-local connections
persisted across tests causing stale DB reads; fixed via conftest teardown and
lifespan db.close() on shutdown.
115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
"""Printer CRUD API — POST /printers, DELETE /printers/{id}."""
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from peewee import JOIN
|
|
|
|
from imptune.db.models import Client, Printer
|
|
|
|
router = APIRouter(prefix="/printers")
|
|
|
|
templates = Jinja2Templates(
|
|
directory=str(Path(__file__).parent.parent / "templates")
|
|
)
|
|
|
|
_VALID_DUPLEX = {"OneSided", "LongEdge", "ShortEdge"}
|
|
_VALID_PAPER = {"A4", "Letter", "Legal"}
|
|
|
|
|
|
def _error_response(message: str, status_code: int = 400) -> HTMLResponse:
|
|
"""Return an HTMX-friendly error fragment swapped into #printer-list."""
|
|
return HTMLResponse(
|
|
content=f"<div id='printer-list' class='error'><p>{message}</p></div>",
|
|
status_code=status_code,
|
|
)
|
|
|
|
|
|
def _render_printer_list(request: Request) -> HTMLResponse:
|
|
"""Query printers with LEFT JOIN on client and render grouped partial."""
|
|
query = (
|
|
Printer.select(Printer, Client)
|
|
.join(Client, JOIN.LEFT_OUTER)
|
|
.order_by(Client.name, Printer.name)
|
|
)
|
|
grouped: dict[str, list[Printer]] = defaultdict(list)
|
|
for p in query:
|
|
client_name = p.client.name if p.client_id else "Unassigned"
|
|
grouped[client_name].append(p)
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="partials/printer_list.html",
|
|
context={"grouped": grouped},
|
|
)
|
|
|
|
|
|
@router.post("", response_class=HTMLResponse)
|
|
def create_printer(
|
|
request: Request,
|
|
name: str = Form(...),
|
|
ip_address: str = Form(...),
|
|
port_name: str = Form(...),
|
|
duplex_mode: str = Form("OneSided"),
|
|
color_mode: str = Form(""),
|
|
paper_size: str = Form("A4"),
|
|
collate: str = Form(""),
|
|
client_id: str = Form(""),
|
|
driver_id: str = Form(""),
|
|
) -> HTMLResponse:
|
|
"""Create a new printer configuration.
|
|
|
|
Boolean fields (color_mode, collate) use HTML checkbox convention:
|
|
"on" = True, absent/empty = False.
|
|
"""
|
|
name = name.strip()
|
|
ip_address = ip_address.strip()
|
|
port_name = port_name.strip()
|
|
|
|
if not name:
|
|
return _error_response("Printer name is required.")
|
|
if not ip_address:
|
|
return _error_response("IP address is required.")
|
|
if not port_name:
|
|
return _error_response("Port name is required.")
|
|
if duplex_mode not in _VALID_DUPLEX:
|
|
return _error_response(f"Invalid duplex mode: {duplex_mode}.")
|
|
if paper_size not in _VALID_PAPER:
|
|
return _error_response(f"Invalid paper size: {paper_size}.")
|
|
|
|
# Convert checkbox values
|
|
color_mode_bool = color_mode == "on"
|
|
collate_bool = collate == "on"
|
|
|
|
# Resolve optional FK IDs
|
|
client_fk = int(client_id) if client_id.strip() else None
|
|
driver_fk = int(driver_id) if driver_id.strip() else None
|
|
|
|
Printer.create(
|
|
name=name,
|
|
ip_address=ip_address,
|
|
port_name=port_name,
|
|
duplex_mode=duplex_mode,
|
|
color_mode=color_mode_bool,
|
|
paper_size=paper_size,
|
|
collate=collate_bool,
|
|
client=client_fk,
|
|
driver=driver_fk,
|
|
)
|
|
|
|
return _render_printer_list(request)
|
|
|
|
|
|
@router.delete("/{printer_id}", response_class=HTMLResponse)
|
|
def delete_printer(request: Request, printer_id: int) -> HTMLResponse:
|
|
"""Delete a printer by ID. Returns updated printer list partial."""
|
|
deleted = Printer.delete().where(Printer.id == printer_id).execute()
|
|
if not deleted:
|
|
return _error_response(f"Printer {printer_id} not found.", status_code=404)
|
|
|
|
return _render_printer_list(request)
|