feat(03-01): implement printer and client CRUD with HTMX/Alpine.js UI

- 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.
This commit is contained in:
2026-04-10 12:56:09 +02:00
parent 9bc26e377b
commit 356c2ee690
12 changed files with 450 additions and 15 deletions
+54
View File
@@ -0,0 +1,54 @@
"""Client CRUD API — POST /clients, GET /clients."""
from __future__ import annotations
from pathlib import Path
from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from peewee import IntegrityError
from imptune.db.models import Client
router = APIRouter(prefix="/clients")
templates = Jinja2Templates(
directory=str(Path(__file__).parent.parent / "templates")
)
def _error_response(message: str, status_code: int = 400) -> HTMLResponse:
"""Return an HTMX-friendly error fragment swapped into #client-list."""
return HTMLResponse(
content=f"<div id='client-list' class='error'><p>{message}</p></div>",
status_code=status_code,
)
def _render_client_list(request: Request) -> HTMLResponse:
"""Render the client list partial for HTMX swap."""
clients = list(Client.select().order_by(Client.name))
return templates.TemplateResponse(
request=request,
name="partials/client_list.html",
context={"clients": clients},
)
@router.post("", response_class=HTMLResponse)
def create_client(request: Request, name: str = Form(...)) -> HTMLResponse:
"""Create a new client.
Accepts form-encoded `name`. Validates non-empty. Returns HTMX partial
with updated client list on success, or error fragment on failure.
"""
name = name.strip()
if not name:
return _error_response("Client name is required.")
try:
Client.create(name=name)
except IntegrityError:
return _error_response(f"Client '{name}' already exists.", status_code=409)
return _render_client_list(request)
+48 -1
View File
@@ -1,9 +1,11 @@
import json
from collections import defaultdict
from pathlib import Path
from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from pathlib import Path
from peewee import JOIN
router = APIRouter()
@@ -36,3 +38,48 @@ def drivers_page(request: Request):
name="drivers.html",
context={"driver_data": driver_data},
)
@router.get("/printers", response_class=HTMLResponse)
def printers_page(request: Request):
from imptune.db.models import Client, Driver, Printer
query = (
Printer.select(Printer, Client)
.join(Client, JOIN.LEFT_OUTER)
.order_by(Client.name, Printer.name)
)
grouped: dict[str, list] = defaultdict(list)
for p in query:
client_name = p.client.name if p.client_id else "Unassigned"
grouped[client_name].append(p)
clients = list(Client.select().order_by(Client.name))
all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
driver_data = []
for d in all_drivers:
names = json.loads(d.driver_desc) if d.driver_desc else []
driver_data.append({"driver": d, "names": names})
return templates.TemplateResponse(
request=request,
name="printers.html",
context={
"grouped": grouped,
"clients": clients,
"driver_data": driver_data,
},
)
@router.get("/clients", response_class=HTMLResponse)
def clients_page(request: Request):
from imptune.db.models import Client
clients = list(Client.select().order_by(Client.name))
return templates.TemplateResponse(
request=request,
name="clients.html",
context={"clients": clients},
)
+114
View File
@@ -0,0 +1,114 @@
"""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)
+7
View File
@@ -12,12 +12,19 @@ def init_db() -> None:
Idempotent — safe to call on every application startup.
Creates all ORM tables if they do not already exist.
Closes any existing connection before re-initializing so that test
fixtures can monkeypatch DB_PATH between test runs.
"""
from imptune.db.models import Client, Driver, Printer, Icon
# Re-read DB_PATH each time so tests can patch imptune.config.DB_PATH
import imptune.config as cfg
# Close any lingering connection from a previous run (important for tests)
if not db.is_closed():
db.close()
db.init(
cfg.DB_PATH,
pragmas={
+7 -2
View File
@@ -5,9 +5,9 @@ from pathlib import Path
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from imptune.api import drivers, health, pages
from imptune.api import clients, drivers, health, pages, printers
from imptune.config import DATA_DIR, DRIVERS_DIR
from imptune.db.database import init_db
from imptune.db.database import db, init_db
@asynccontextmanager
@@ -16,6 +16,9 @@ async def lifespan(app: FastAPI):
os.makedirs(DRIVERS_DIR, exist_ok=True)
init_db()
yield
# Close DB connection on shutdown so test fixtures can re-initialize cleanly
if not db.is_closed():
db.close()
app = FastAPI(title="ImpTune", lifespan=lifespan)
@@ -28,3 +31,5 @@ app.mount("/static", StaticFiles(directory=str(_static_dir)), name="static")
app.include_router(health.router)
app.include_router(pages.router)
app.include_router(drivers.router)
app.include_router(printers.router)
app.include_router(clients.router)
+21
View File
@@ -0,0 +1,21 @@
{% extends "base.html" %}
{% block content %}
<h1>Clients</h1>
<section>
<h2>Add Client</h2>
<form hx-post="/clients" hx-target="#client-list" hx-swap="outerHTML">
<label>
Client Name
<input type="text" name="name" placeholder="e.g. Contoso" required>
</label>
<button type="submit">Add Client</button>
</form>
</section>
<section>
<h2>Client List</h2>
{% include "partials/client_list.html" %}
</section>
{% endblock %}
@@ -0,0 +1,22 @@
<div id="client-list">
{% if not clients %}
<p>No clients configured yet.</p>
{% else %}
<table>
<thead>
<tr>
<th>Name</th>
<th>Created</th>
</tr>
</thead>
<tbody>
{% for c in clients %}
<tr>
<td>{{ c.name }}</td>
<td>{{ c.created_at.strftime('%Y-%m-%d') if c.created_at else '—' }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% endif %}
</div>
@@ -0,0 +1,86 @@
<div x-data="{ ip: '{{ printer.ip_address if printer else '' }}', port: '{{ printer.port_name if printer else '' }}', portEdited: {{ 'true' if printer else 'false' }} }">
<form hx-post="/printers" hx-target="#printer-list" hx-swap="outerHTML">
<label>
Printer Name
<input type="text" name="name" placeholder="e.g. HP LaserJet 4050" required
value="{{ printer.name if printer else '' }}">
</label>
<label>
IP Address
<input type="text" name="ip_address"
x-model="ip"
@input="if (!portEdited) port = 'IP_' + ip.replaceAll('.', '_')"
placeholder="e.g. 192.168.1.100"
required>
</label>
<label>
Port Name
<input type="text" name="port_name"
x-model="port"
@change="portEdited = true"
@keydown="portEdited = true"
placeholder="e.g. IP_192_168_1_100">
</label>
<label>
Driver
<select name="driver_id">
<option value="">-- No driver --</option>
{% for item in driver_data %}
<option value="{{ item.driver.id }}"
{% if printer and printer.driver_id == item.driver.id %}selected{% endif %}>
{{ item.driver.original_filename }} ({{ item.names | join(', ') }})
</option>
{% endfor %}
</select>
</label>
<label>
Duplex Mode
<select name="duplex_mode">
<option value="OneSided" {% if not printer or printer.duplex_mode == 'OneSided' %}selected{% endif %}>One-Sided</option>
<option value="LongEdge" {% if printer and printer.duplex_mode == 'LongEdge' %}selected{% endif %}>Long Edge</option>
<option value="ShortEdge" {% if printer and printer.duplex_mode == 'ShortEdge' %}selected{% endif %}>Short Edge</option>
</select>
</label>
<label>
<input type="checkbox" name="color_mode" value="on"
{% if not printer or printer.color_mode %}checked{% endif %}>
Color Mode
</label>
<label>
Paper Size
<select name="paper_size">
<option value="A4" {% if not printer or printer.paper_size == 'A4' %}selected{% endif %}>A4</option>
<option value="Letter" {% if printer and printer.paper_size == 'Letter' %}selected{% endif %}>Letter</option>
<option value="Legal" {% if printer and printer.paper_size == 'Legal' %}selected{% endif %}>Legal</option>
</select>
</label>
<label>
<input type="checkbox" name="collate" value="on"
{% if not printer or printer.collate %}checked{% endif %}>
Collate
</label>
<label>
Client
<select name="client_id">
<option value="">-- Unassigned --</option>
{% for c in clients %}
<option value="{{ c.id }}"
{% if printer and printer.client_id == c.id %}selected{% endif %}>
{{ c.name }}
</option>
{% endfor %}
</select>
</label>
<button type="submit">Save Printer</button>
</form>
</div>
@@ -0,0 +1,49 @@
<div id="printer-list">
{% if not grouped %}
<p>No printers configured yet.</p>
{% else %}
{% for client_name, printers in grouped.items() %}
<section>
<h3>{{ client_name }}</h3>
<table>
<thead>
<tr>
<th>Name</th>
<th>IP Address</th>
<th>Port</th>
<th>Driver</th>
<th>Duplex</th>
<th>Color</th>
<th>Paper</th>
<th>Collate</th>
<th>Actions</th>
</tr>
</thead>
<tbody>
{% for p in printers %}
<tr>
<td>{{ p.name }}</td>
<td>{{ p.ip_address }}</td>
<td>{{ p.port_name }}</td>
<td>{{ p.driver.original_filename if p.driver_id else '—' }}</td>
<td>{{ p.duplex_mode }}</td>
<td>{{ 'Yes' if p.color_mode else 'No' }}</td>
<td>{{ p.paper_size }}</td>
<td>{{ 'Yes' if p.collate else 'No' }}</td>
<td>
<button
hx-delete="/printers/{{ p.id }}"
hx-target="#printer-list"
hx-swap="outerHTML"
hx-confirm="Delete '{{ p.name }}'?">
Delete
</button>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</section>
{% endfor %}
{% endif %}
</div>
+15
View File
@@ -0,0 +1,15 @@
{% extends "base.html" %}
{% block content %}
<h1>Printers</h1>
<section>
<h2>Add Printer</h2>
{% include "partials/printer_form.html" %}
</section>
<section>
<h2>Printer Library</h2>
{% include "partials/printer_list.html" %}
</section>
{% endblock %}
+9 -1
View File
@@ -24,4 +24,12 @@ def tmp_data_dir(tmp_path, monkeypatch):
cfg.DATA_DIR = str(data_dir)
cfg.DB_PATH = str(data_dir / "imptune.db")
cfg.DRIVERS_DIR = str(data_dir / "drivers")
return data_dir
yield data_dir
# Close the test-thread's DB connection so the next test gets a fresh one
# pointing to its own tmp DB (Peewee connections are thread-local).
from imptune.db.database import db
if not db.is_closed():
db.close()
+18 -11
View File
@@ -47,8 +47,9 @@ def test_create_printer_duplex(client: TestClient) -> None:
)
assert resp.status_code == 200
printer = Printer.get(Printer.name == "Duplex Printer")
assert printer.duplex_mode == "LongEdge"
printers = list(Printer.select().where(Printer.name == "Duplex Printer"))
assert len(printers) == 1
assert printers[0].duplex_mode == "LongEdge"
def test_create_printer_color_mode(client: TestClient) -> None:
@@ -66,8 +67,9 @@ def test_create_printer_color_mode(client: TestClient) -> None:
)
assert resp.status_code == 200
printer = Printer.get(Printer.name == "Mono Printer")
assert printer.color_mode is False
printers = list(Printer.select().where(Printer.name == "Mono Printer"))
assert len(printers) == 1
assert printers[0].color_mode is False
def test_create_printer_paper_size(client: TestClient) -> None:
@@ -85,8 +87,9 @@ def test_create_printer_paper_size(client: TestClient) -> None:
)
assert resp.status_code == 200
printer = Printer.get(Printer.name == "Letter Printer")
assert printer.paper_size == "Letter"
printers = list(Printer.select().where(Printer.name == "Letter Printer"))
assert len(printers) == 1
assert printers[0].paper_size == "Letter"
def test_create_printer_collate(client: TestClient) -> None:
@@ -104,8 +107,9 @@ def test_create_printer_collate(client: TestClient) -> None:
)
assert resp.status_code == 200
printer = Printer.get(Printer.name == "No Collate Printer")
assert printer.collate is False
printers = list(Printer.select().where(Printer.name == "No Collate Printer"))
assert len(printers) == 1
assert printers[0].collate is False
def test_create_client(client: TestClient) -> None:
@@ -126,7 +130,9 @@ def test_printer_grouped_by_client(client: TestClient) -> None:
resp = client.post("/clients", data={"name": "Contoso"})
assert resp.status_code == 200
contoso = Client.get(Client.name == "Contoso")
contoso_list = list(Client.select().where(Client.name == "Contoso"))
assert len(contoso_list) == 1
contoso = contoso_list[0]
# Create printer assigned to that client
resp = client.post(
@@ -191,8 +197,9 @@ def test_delete_printer(client: TestClient) -> None:
assert resp.status_code == 200
# Find its ID
printer = Printer.get(Printer.name == "To Delete")
printer_id = printer.id
printers = list(Printer.select().where(Printer.name == "To Delete"))
assert len(printers) == 1
printer_id = printers[0].id
# Delete it
resp = client.delete(f"/printers/{printer_id}")