From 356c2ee6904c78fa640464116e132233ff235d00 Mon Sep 17 00:00:00 2001 From: Kawa Date: Fri, 10 Apr 2026 12:56:09 +0200 Subject: [PATCH] feat(03-01): implement printer and client CRUD with HTMX/Alpine.js UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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. --- imptune/api/clients.py | 54 +++++++++ imptune/api/pages.py | 49 +++++++- imptune/api/printers.py | 114 +++++++++++++++++++ imptune/db/database.py | 7 ++ imptune/main.py | 9 +- imptune/templates/clients.html | 21 ++++ imptune/templates/partials/client_list.html | 22 ++++ imptune/templates/partials/printer_form.html | 86 ++++++++++++++ imptune/templates/partials/printer_list.html | 49 ++++++++ imptune/templates/printers.html | 15 +++ tests/conftest.py | 10 +- tests/test_printer_crud.py | 29 +++-- 12 files changed, 450 insertions(+), 15 deletions(-) create mode 100644 imptune/api/clients.py create mode 100644 imptune/api/printers.py create mode 100644 imptune/templates/clients.html create mode 100644 imptune/templates/partials/client_list.html create mode 100644 imptune/templates/partials/printer_form.html create mode 100644 imptune/templates/partials/printer_list.html create mode 100644 imptune/templates/printers.html diff --git a/imptune/api/clients.py b/imptune/api/clients.py new file mode 100644 index 0000000..db40556 --- /dev/null +++ b/imptune/api/clients.py @@ -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"

{message}

", + 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) diff --git a/imptune/api/pages.py b/imptune/api/pages.py index 814f20a..7d6de82 100644 --- a/imptune/api/pages.py +++ b/imptune/api/pages.py @@ -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}, + ) diff --git a/imptune/api/printers.py b/imptune/api/printers.py new file mode 100644 index 0000000..0af9203 --- /dev/null +++ b/imptune/api/printers.py @@ -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"

{message}

", + 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) diff --git a/imptune/db/database.py b/imptune/db/database.py index 999979a..b3d19a1 100644 --- a/imptune/db/database.py +++ b/imptune/db/database.py @@ -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={ diff --git a/imptune/main.py b/imptune/main.py index ab57d63..b093308 100644 --- a/imptune/main.py +++ b/imptune/main.py @@ -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) diff --git a/imptune/templates/clients.html b/imptune/templates/clients.html new file mode 100644 index 0000000..051637b --- /dev/null +++ b/imptune/templates/clients.html @@ -0,0 +1,21 @@ +{% extends "base.html" %} + +{% block content %} +

Clients

+ +
+

Add Client

+
+ + +
+
+ +
+

Client List

+ {% include "partials/client_list.html" %} +
+{% endblock %} diff --git a/imptune/templates/partials/client_list.html b/imptune/templates/partials/client_list.html new file mode 100644 index 0000000..4e03dc8 --- /dev/null +++ b/imptune/templates/partials/client_list.html @@ -0,0 +1,22 @@ +
+ {% if not clients %} +

No clients configured yet.

+ {% else %} + + + + + + + + + {% for c in clients %} + + + + + {% endfor %} + +
NameCreated
{{ c.name }}{{ c.created_at.strftime('%Y-%m-%d') if c.created_at else '—' }}
+ {% endif %} +
diff --git a/imptune/templates/partials/printer_form.html b/imptune/templates/partials/printer_form.html new file mode 100644 index 0000000..e0dde80 --- /dev/null +++ b/imptune/templates/partials/printer_form.html @@ -0,0 +1,86 @@ +
+
+ + + + + + + + + + + + + + + + + + + + +
+
diff --git a/imptune/templates/partials/printer_list.html b/imptune/templates/partials/printer_list.html new file mode 100644 index 0000000..f9f6edf --- /dev/null +++ b/imptune/templates/partials/printer_list.html @@ -0,0 +1,49 @@ +
+ {% if not grouped %} +

No printers configured yet.

+ {% else %} + {% for client_name, printers in grouped.items() %} +
+

{{ client_name }}

+ + + + + + + + + + + + + + + + {% for p in printers %} + + + + + + + + + + + + {% endfor %} + +
NameIP AddressPortDriverDuplexColorPaperCollateActions
{{ p.name }}{{ p.ip_address }}{{ p.port_name }}{{ p.driver.original_filename if p.driver_id else '—' }}{{ p.duplex_mode }}{{ 'Yes' if p.color_mode else 'No' }}{{ p.paper_size }}{{ 'Yes' if p.collate else 'No' }} + +
+
+ {% endfor %} + {% endif %} +
diff --git a/imptune/templates/printers.html b/imptune/templates/printers.html new file mode 100644 index 0000000..0819b28 --- /dev/null +++ b/imptune/templates/printers.html @@ -0,0 +1,15 @@ +{% extends "base.html" %} + +{% block content %} +

Printers

+ +
+

Add Printer

+ {% include "partials/printer_form.html" %} +
+ +
+

Printer Library

+ {% include "partials/printer_list.html" %} +
+{% endblock %} diff --git a/tests/conftest.py b/tests/conftest.py index de08e3d..5d955ae 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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() diff --git a/tests/test_printer_crud.py b/tests/test_printer_crud.py index 20044bd..c194780 100644 --- a/tests/test_printer_crud.py +++ b/tests/test_printer_crud.py @@ -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}")