"""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, RedirectResponse
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"
",
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 RedirectResponse(url="/printers", status_code=303)
@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)