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:
@@ -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
@@ -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},
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
@@ -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
@@ -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)
|
||||
|
||||
@@ -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>
|
||||
@@ -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
@@ -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
@@ -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}")
|
||||
|
||||
Reference in New Issue
Block a user