feat: memory-only sessions on HTTP, streamed exports, UI refresh

Session

- COOKIE_SECURE=false no longer persists the owner key for ten years.
  services/session.cookie_kwargs() drops max_age in that mode, so the
  browser holds the key in memory and the session ends with the window.
  Everything still persists server-side; only the browser link is
  temporary. base.html shows a warning banner (FR/EN) and an extra
  paragraph in the onboarding modal, and the README explains the
  trade-off and the backup-key escape hatch.
- Both cookie writers (middleware, POST /session/restore) go through
  cookie_kwargs() so the policy cannot drift between them.
- The CSRF guard on /session/restore compared request.url.scheme against
  the Origin header. Behind a TLS-terminating proxy uvicorn sees http
  while the browser sends https, so every legitimate restore was
  rejected with 403. It now compares hosts only, including
  X-Forwarded-Host.
- /static/*, /favicon.ico and /robots.txt skip the middleware. Each
  cookieless hit was inserting an Owner row no browser could ever use.

Reliability

- Malformed printer-form FK fields no longer escape as HTTP 500:
  a non-numeric client_id/driver_id raised ValueError and an unknown
  driver_id hit a FOREIGN KEY constraint. Both are now 400/404 HTMX
  fragments, and the duplicated field checks moved into
  _validate_fields().
- Package exports stream. build_intunewin() encrypts the inner ZIP in
  1 MB chunks against temp files with a streaming HMAC and SHA256, and
  both endpoints serve the result with FileResponse plus a background
  cleanup task. A 100 MB driver used to be held in memory three or four
  times over per concurrent download. The byte layout is unchanged.
- FileResponse also escapes the download filename, which was previously
  interpolated raw into Content-Disposition.
- python-multipart >= 0.0.18 (CVE-2024-53981, reachable from
  /drivers/upload) and Pillow >= 10.3 (CVE-2024-28219, reachable from
  icon upload).
- icons.py reads cfg.ICONS_DIR instead of re-deriving the path from
  DATA_DIR, matching the .intunewin export.

UI

- Sidebar/topbar shell, inline SVG icon macros (partials/icons.html),
  card and data-table components, grouped printer list, and the
  dedicated /printers/new page replacing partials/printer_form.html.

Tests

- 194 pass with a bare `pytest tests/`: tests/conftest.py now forces
  cfg.COOKIE_SECURE = False like the e2e conftest already did, so the
  Secure cookie is no longer dropped over http://testserver.
- New coverage for the malformed-FK guards, the chunk-boundary cases in
  the encrypt loop (every residue mod _CHUNK plus a multi-megabyte
  payload), temp-dir cleanup after both exports, and the whole
  COOKIE_SECURE matrix.
- test_printer_edit.py located the Edit button by its translated label,
  so it only passed on English-locale machines. It now targets the
  showModal() hook, which also cuts the e2e run from 84s to 15s.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

imptune/api/clients.py

@@ -27,11 +27,15 @@ def _error_response(message: str, status_code: int = 400) -> HTMLResponse:
 
 def _render_client_list(request: Request) -> HTMLResponse:
     """Render the client list partial for HTMX swap."""
-    clients = list(Client.select().where(Client.owner == request.state.owner).order_by(Client.name))
+    from imptune.api.pages import printer_counts_by_client
+
+    owner = request.state.owner
+    clients = list(Client.select().where(Client.owner == owner).order_by(Client.name))
     return templates.TemplateResponse(
         request=request,
         name="partials/client_list.html",
-        context={"clients": clients},
+        # Counts must come along, or the swapped-in rows all read "0 printers".
+        context={"clients": clients, "counts": printer_counts_by_client(owner)},
     )
 
 

imptune/api/icons.py

@@ -6,7 +6,7 @@ import io
 from pathlib import Path
 
 from fastapi import APIRouter, Request, UploadFile
-from fastapi.responses import HTMLResponse
+from fastapi.responses import FileResponse, HTMLResponse, Response
 from PIL import Image
 
 import imptune.config as cfg
@@ -68,9 +68,11 @@ def upload_icon(request: Request, printer_id: int, file: UploadFile) -> HTMLResp
             status_code=422,
         )
 
-    # Store SHA256-addressed on disk
+    # Store SHA256-addressed on disk. cfg.ICONS_DIR is the same path the
+    # .intunewin export reads from — deriving it a second time from DATA_DIR
+    # would silently diverge if the layout ever changes.
     sha256 = hashlib.sha256(data).hexdigest()
-    icons_dir = Path(cfg.DATA_DIR) / "icons"
+    icons_dir = Path(cfg.ICONS_DIR)
     icons_dir.mkdir(parents=True, exist_ok=True)
     icon_path = icons_dir / sha256
     icon_path.write_bytes(data)
@@ -85,6 +87,37 @@ def upload_icon(request: Request, printer_id: int, file: UploadFile) -> HTMLResp
     )
 
     return HTMLResponse(
-        content="<p>Icon uploaded successfully</p>",
+        content=(
+            '<p class="ok-note">Icon uploaded successfully</p>'
+            f'<div class="icon-preview"><img src="/printers/{printer_id}/icon?v={sha256[:8]}"'
+            f' width="56" height="56" alt=""><span class="meta">{img.size[0]}&times;{img.size[1]} PNG</span></div>'
+        ),
         status_code=200,
     )
+
+
+@router.get("/{printer_id}/icon")
+def get_icon(request: Request, printer_id: int) -> Response:
+    """Serve the stored 256x256 PNG so the UI can show what was uploaded.
+
+    Owner-scoped: a printer belonging to another owner reads as missing.
+    """
+    printer = Printer.get_or_none(
+        (Printer.id == printer_id) & (Printer.owner == request.state.owner)
+    )
+    if printer is None:
+        return Response(status_code=404)
+
+    icon = Icon.get_or_none(Icon.printer == printer_id)
+    if icon is None:
+        return Response(status_code=404)
+
+    icon_path = Path(cfg.ICONS_DIR) / icon.sha256
+    if not icon_path.exists():
+        return Response(status_code=404)
+
+    return FileResponse(
+        icon_path,
+        media_type="image/png",
+        headers={"Cache-Control": "private, max-age=300"},
+    )

imptune/api/packages.py

@@ -1,5 +1,4 @@
 """Package export endpoints — serves deployment packages for NinjaRMM and Microsoft Intune."""
-import io
 import json
 import os
 import shutil
@@ -7,7 +6,8 @@ import tempfile
 import zipfile
 
 from fastapi import APIRouter, Request
-from fastapi.responses import PlainTextResponse, Response
+from fastapi.responses import FileResponse, PlainTextResponse
+from starlette.background import BackgroundTask
 
 import imptune.config as cfg
 from imptune.db.models import Icon, Owner, Printer
@@ -17,6 +17,11 @@ from imptune.storage.driver_store import DriverStore
 
 router = APIRouter(prefix="/printers")
 
+# Driver payloads run to ~100 MB. Packages are assembled in a temp dir and
+# streamed from disk instead of being held in memory, so N concurrent
+# downloads cost N file handles rather than N × package size of RAM.
+_CHUNK = 1024 * 1024
+
 
 def _get_printer_and_driver(printer_id: int, owner: Owner):
     """Fetch printer (scoped to owner) and validate driver — returns (printer, driver, driver_name) or PlainTextResponse error."""
@@ -47,6 +52,16 @@ def _get_driver_zip_path(driver) -> str:
     return str(DriverStore(cfg.DRIVERS_DIR).get_path(driver.sha256))
 
 
+def _streamed_download(path: str, tmpdir: str, media_type: str, filename: str) -> FileResponse:
+    """Serve a built package off disk, deleting its temp dir once sent."""
+    return FileResponse(
+        path,
+        media_type=media_type,
+        filename=filename,
+        background=BackgroundTask(shutil.rmtree, tmpdir, True),
+    )
+
+
 @router.get("/{printer_id}/packages/ninja")
 def get_ninja_package(request: Request, printer_id: int):
     """Download a NinjaRMM-ready ZIP containing install.ps1 and the driver files."""
@@ -76,22 +91,30 @@ def get_ninja_package(request: Request, printer_id: int):
         collate=printer.collate,
     )
 
-    # Build ZIP in-memory
-    buf = io.BytesIO()
-    with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
-        # Add install script
-        zf.writestr(f"{safe_name}/install.ps1", install_script)
+    # Build the ZIP on disk, copying driver members through in chunks so a
+    # 100 MB driver never lands in memory whole.
+    tmpdir = tempfile.mkdtemp(prefix="imptune_ninja_")
+    try:
+        # Fixed on-disk name — the printer name only shapes the download name,
+        # so a "/" or ":" in it can't break the temp path.
+        out_path = os.path.join(tmpdir, "package.zip")
+        with zipfile.ZipFile(out_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
+            zf.writestr(f"{safe_name}/install.ps1", install_script)
 
-        # Extract and re-add driver files from driver ZIP
-        with zipfile.ZipFile(driver_zip_path, "r") as driver_zf:
-            for member in driver_zf.namelist():
-                member_data = driver_zf.read(member)
-                zf.writestr(f"{safe_name}/drivers/{member}", member_data)
+            with zipfile.ZipFile(driver_zip_path, "r") as driver_zf:
+                for member in driver_zf.infolist():
+                    target = f"{safe_name}/drivers/{member.filename}"
+                    if member.is_dir():
+                        zf.writestr(target, b"")
+                        continue
+                    with driver_zf.open(member) as src, zf.open(target, "w") as dst:
+                        shutil.copyfileobj(src, dst, _CHUNK)
+    except BaseException:
+        shutil.rmtree(tmpdir, ignore_errors=True)
+        raise
 
-    return Response(
-        content=buf.getvalue(),
-        media_type="application/zip",
-        headers={"Content-Disposition": f'attachment; filename="{safe_name}_ninja.zip"'},
+    return _streamed_download(
+        out_path, tmpdir, "application/zip", f"{safe_name}_ninja.zip"
     )
 
 
@@ -130,17 +153,24 @@ def get_intunewin_package(request: Request, printer_id: int):
     )
     detect_script = render_detect(printer_name=printer.name)
 
-    with tempfile.TemporaryDirectory(prefix="imptune_") as tmpdir:
-        # Write scripts
-        with open(os.path.join(tmpdir, "install.ps1"), "w", encoding="utf-8") as f:
-            f.write(install_script)
-        with open(os.path.join(tmpdir, "uninstall.ps1"), "w", encoding="utf-8") as f:
-            f.write(uninstall_script)
-        with open(os.path.join(tmpdir, "detect.ps1"), "w", encoding="utf-8") as f:
-            f.write(detect_script)
+    # The build output is streamed straight off disk, so the temp tree has to
+    # outlive this handler — the response's background task removes it.
+    tmpdir = tempfile.mkdtemp(prefix="imptune_")
+    try:
+        staging = os.path.join(tmpdir, "staging")
+        os.makedirs(staging, exist_ok=True)
 
-        # Extract driver ZIP contents into tmpdir/drivers/
-        drivers_subdir = os.path.join(tmpdir, "drivers")
+        # Write scripts
+        for filename, script in (
+            ("install.ps1", install_script),
+            ("uninstall.ps1", uninstall_script),
+            ("detect.ps1", detect_script),
+        ):
+            with open(os.path.join(staging, filename), "w", encoding="utf-8") as f:
+                f.write(script)
+
+        # Extract driver ZIP contents into staging/drivers/
+        drivers_subdir = os.path.join(staging, "drivers")
         os.makedirs(drivers_subdir, exist_ok=True)
         with zipfile.ZipFile(driver_zip_path, "r") as driver_zf:
             driver_zf.extractall(drivers_subdir)
@@ -150,17 +180,16 @@ def get_intunewin_package(request: Request, printer_id: int):
         if icon_record is not None:
             icon_src = os.path.join(cfg.ICONS_DIR, icon_record.sha256)
             if os.path.isfile(icon_src):
-                shutil.copy2(icon_src, os.path.join(tmpdir, "icon.png"))
+                shutil.copy2(icon_src, os.path.join(staging, "icon.png"))
 
-        # Build .intunewin
-        output_path = os.path.join(tmpdir, "out.intunewin")
-        build_intunewin(tmpdir, "install.ps1", output_path)
+        # Build .intunewin outside the staging dir — an output file written into
+        # the tree being packaged would end up inside its own package.
+        output_path = os.path.join(tmpdir, "package.intunewin")
+        build_intunewin(staging, "install.ps1", output_path)
+    except BaseException:
+        shutil.rmtree(tmpdir, ignore_errors=True)
+        raise
 
-        with open(output_path, "rb") as f:
-            content = f.read()
-
-    return Response(
-        content=content,
-        media_type="application/octet-stream",
-        headers={"Content-Disposition": f'attachment; filename="{safe_name}.intunewin"'},
+    return _streamed_download(
+        output_path, tmpdir, "application/octet-stream", f"{safe_name}.intunewin"
     )

imptune/api/pages.py

@@ -5,37 +5,102 @@ from pathlib import Path
 from fastapi import APIRouter, Request
 from fastapi.responses import HTMLResponse
 from fastapi.templating import Jinja2Templates
-from peewee import JOIN
+from peewee import JOIN, fn
 
 
 router = APIRouter()
 
 templates = Jinja2Templates(directory=str(Path(__file__).parent.parent / "templates"))
 
+# Group label for printers with no client — the template translates it.
+UNASSIGNED_GROUP = "Unassigned"
+
+
+def printer_counts_by_driver(owner) -> dict[int, int]:
+    """How many of this owner's printers use each (globally shared) driver."""
+    from imptune.db.models import Printer
+
+    rows = (
+        Printer.select(Printer.driver, fn.COUNT(Printer.id).alias("n"))
+        .where((Printer.owner == owner) & Printer.driver.is_null(False))
+        .group_by(Printer.driver)
+    )
+    return {row.driver_id: row.n for row in rows}
+
+
+def printer_counts_by_client(owner) -> dict[int, int]:
+    """How many printers each client groups."""
+    from imptune.db.models import Printer
+
+    rows = (
+        Printer.select(Printer.client, fn.COUNT(Printer.id).alias("n"))
+        .where((Printer.owner == owner) & Printer.client.is_null(False))
+        .group_by(Printer.client)
+    )
+    return {row.client_id: row.n for row in rows}
+
+
+def group_printers_by_client(printers) -> dict[str, list]:
+    """Group printers under their client name, unassigned ones last.
+
+    The template renders groups in insertion order, and "a printer nobody has
+    filed yet" belongs at the bottom of the page, not the top.
+    """
+    grouped: dict[str, list] = defaultdict(list)
+    for p in printers:
+        grouped[p.client.name if p.client_id else UNASSIGNED_GROUP].append(p)
+
+    unassigned = grouped.pop(UNASSIGNED_GROUP, None)
+    ordered = {name: grouped[name] for name in sorted(grouped)}
+    if unassigned:
+        ordered[UNASSIGNED_GROUP] = unassigned
+    return ordered
+
+
+def _driver_data(drivers) -> list[dict]:
+    """Attach the parsed driver names to each Driver row for template use."""
+    return [
+        {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []}
+        for d in drivers
+    ]
+
 
 @router.get("/", response_class=HTMLResponse)
 def dashboard(request: Request):
-    from imptune.db.models import Printer
+    from imptune.db.models import Client, Driver, Printer
 
     owner = request.state.owner
     recent_printers = list(
-        Printer.select()
+        Printer.select(Printer, Client)
+        .join(Client, JOIN.LEFT_OUTER)
         .where(Printer.owner == owner)
         .order_by(Printer.created_at.desc())
         .limit(5)
     )
     recent_packages = list(
-        Printer.select()
+        Printer.select(Printer, Client)
+        .join(Client, JOIN.LEFT_OUTER)
+        .switch(Printer)
         .where((Printer.owner == owner) & Printer.driver.is_null(False))
         .order_by(Printer.created_at.desc())
         .limit(5)
     )
+    printer_count = Printer.select().where(Printer.owner == owner).count()
+    ready_count = (
+        Printer.select()
+        .where((Printer.owner == owner) & Printer.driver.is_null(False))
+        .count()
+    )
     return templates.TemplateResponse(
         request=request,
         name="dashboard.html",
         context={
             "recent_printers": recent_printers,
             "recent_packages": recent_packages,
+            "driver_count": Driver.select().count(),
+            "printer_count": printer_count,
+            "client_count": Client.select().where(Client.owner == owner).count(),
+            "ready_count": ready_count,
         },
     )
 
@@ -45,14 +110,13 @@ def drivers_page(request: Request):
     from imptune.db.models import Driver
 
     drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
-    driver_data = []
-    for d in 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="drivers.html",
-        context={"driver_data": driver_data},
+        context={
+            "driver_data": _driver_data(drivers),
+            "usage": printer_counts_by_driver(request.state.owner),
+        },
     )
 
 
@@ -67,18 +131,13 @@ def printers_page(request: Request):
         .where(Printer.owner == owner)
         .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)
+    grouped = group_printers_by_client(query)
 
     clients = list(Client.select().where(Client.owner == owner).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})
+
+    printer_count = sum(len(v) for v in grouped.values())
+    ready_count = sum(1 for v in grouped.values() for p in v if p.driver_id)
 
     return templates.TemplateResponse(
         request=request,
@@ -86,7 +145,9 @@ def printers_page(request: Request):
         context={
             "grouped": grouped,
             "clients": clients,
-            "driver_data": driver_data,
+            "driver_data": _driver_data(all_drivers),
+            "printer_count": printer_count,
+            "ready_count": ready_count,
         },
     )
 
@@ -97,14 +158,10 @@ def printers_new_page(request: Request):
 
     clients = list(Client.select().where(Client.owner == request.state.owner).order_by(Client.name))
     all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
-    driver_data = [
-        {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []}
-        for d in all_drivers
-    ]
     return templates.TemplateResponse(
         request=request,
         name="printers_new.html",
-        context={"clients": clients, "driver_data": driver_data},
+        context={"clients": clients, "driver_data": _driver_data(all_drivers)},
     )
 
 
@@ -154,11 +211,12 @@ def printer_detail(request: Request, printer_id: int):
 def clients_page(request: Request):
     from imptune.db.models import Client
 
-    clients = list(Client.select().where(Client.owner == request.state.owner).order_by(Client.name))
+    owner = request.state.owner
+    clients = list(Client.select().where(Client.owner == owner).order_by(Client.name))
     return templates.TemplateResponse(
         request=request,
         name="clients.html",
-        context={"clients": clients},
+        context={"clients": clients, "counts": printer_counts_by_client(owner)},
     )
 
 
@@ -180,14 +238,11 @@ def client_detail(request: Request, client_id: int):
         .where((Printer.client == client_id) & (Printer.owner == owner))
         .order_by(Printer.name)
     )
-    grouped = {client.name: list(query)}
+    printers = list(query)
+    grouped = {client.name: printers}
 
     clients = list(Client.select().where(Client.owner == owner).order_by(Client.name))
     all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
-    driver_data = [
-        {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []}
-        for d in all_drivers
-    ]
 
     return templates.TemplateResponse(
         request=request,
@@ -196,25 +251,43 @@ def client_detail(request: Request, client_id: int):
             "client": client,
             "grouped": grouped,
             "clients": clients,
-            "driver_data": driver_data,
+            "driver_data": _driver_data(all_drivers),
+            "printer_count": len(printers),
+            "ready_count": sum(1 for p in printers if p.driver_id),
         },
     )
 
 
 @router.get("/packages", response_class=HTMLResponse)
 def packages_page(request: Request):
-    from imptune.db.models import Client, Driver, Printer
+    from imptune.db.models import Client, Driver, Icon, Printer
 
+    owner = request.state.owner
     printers = list(
         Printer.select(Printer, Client, Driver)
         .join(Client, JOIN.LEFT_OUTER)
         .switch(Printer)
         .join(Driver, JOIN.LEFT_OUTER)
-        .where((Printer.owner == request.state.owner) & Printer.driver.is_null(False))
+        .where((Printer.owner == owner) & Printer.driver.is_null(False))
         .order_by(Printer.name)
     )
+    pending_count = (
+        Printer.select()
+        .where((Printer.owner == owner) & Printer.driver.is_null(True))
+        .count()
+    )
+    icon_printer_ids = {
+        row.printer_id
+        for row in Icon.select(Icon.printer).where(
+            Icon.printer.in_([p.id for p in printers] or [0])
+        )
+    }
     return templates.TemplateResponse(
         request=request,
         name="packages.html",
-        context={"printers": printers},
+        context={
+            "printers": printers,
+            "pending_count": pending_count,
+            "icon_printer_ids": icon_printer_ids,
+        },
     )

imptune/api/printers.py

@@ -1,14 +1,14 @@
 """Printer CRUD API — POST /printers, DELETE /printers/{id}, PATCH /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.responses import HTMLResponse, RedirectResponse, Response
 from fastapi.templating import Jinja2Templates
 from peewee import JOIN
 
+from imptune.api.pages import group_printers_by_client
 from imptune.db.models import Client, Driver, Printer
 
 router = APIRouter(prefix="/printers")
@@ -29,6 +29,59 @@ def _error_response(message: str, status_code: int = 400) -> HTMLResponse:
     )
 
 
+def _validate_fields(
+    name: str, ip_address: str, port_name: str, duplex_mode: str, paper_size: str
+) -> HTMLResponse | None:
+    """Shared field validation for create and update — None when everything is valid."""
+    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}.")
+    return None
+
+
+def _resolve_client(raw: str, owner) -> tuple[int | None, HTMLResponse | None]:
+    """Resolve the optional client_id form field to an owned Client id.
+
+    The form value is attacker-controlled text: a non-numeric value used to
+    raise ValueError (HTTP 500) instead of the 400 the HTMX form can render.
+    """
+    raw = raw.strip()
+    if not raw:
+        return None, None
+    try:
+        client_fk = int(raw)
+    except ValueError:
+        return None, _error_response(f"Invalid client id: {raw}.")
+    if Client.get_or_none((Client.id == client_fk) & (Client.owner == owner)) is None:
+        return None, _error_response(f"Client {client_fk} not found.", status_code=404)
+    return client_fk, None
+
+
+def _resolve_driver(raw: str) -> tuple[int | None, HTMLResponse | None]:
+    """Resolve the optional driver_id form field. Drivers are global/shared.
+
+    Existence is checked here because an unknown id otherwise reaches SQLite as
+    a FOREIGN KEY violation — an IntegrityError (HTTP 500) rather than a 404.
+    """
+    raw = raw.strip()
+    if not raw:
+        return None, None
+    try:
+        driver_fk = int(raw)
+    except ValueError:
+        return None, _error_response(f"Invalid driver id: {raw}.")
+    if Driver.get_or_none(Driver.id == driver_fk) is None:
+        return None, _error_response(f"Driver {driver_fk} not found.", status_code=404)
+    return driver_fk, None
+
+
 def _render_printer_list(request: Request) -> HTMLResponse:
     """Query printers with LEFT JOIN on client and render grouped partial."""
     import json
@@ -40,10 +93,7 @@ def _render_printer_list(request: Request) -> HTMLResponse:
         .where(Printer.owner == owner)
         .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)
+    grouped = group_printers_by_client(query)
 
     clients = list(Client.select().where(Client.owner == owner).order_by(Client.name))
     all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
@@ -71,48 +121,39 @@ def create_printer(
     collate: str = Form(""),
     client_id: str = Form(""),
     driver_id: str = Form(""),
-) -> HTMLResponse:
+) -> Response:
     """Create a new printer configuration.
 
-    Boolean fields (color_mode, collate) use HTML checkbox convention:
-    "on" = True, absent/empty = False.
+    Redirects to /printers on success; returns an inline HTMX error fragment
+    (400/404) on validation failure.
     """
     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"
+    invalid = _validate_fields(name, ip_address, port_name, duplex_mode, paper_size)
+    if invalid is not None:
+        return invalid
 
     owner = request.state.owner
 
     # Resolve optional FK IDs — client must belong to this owner
-    client_fk = int(client_id) if client_id.strip() else None
-    if client_fk is not None:
-        if Client.get_or_none((Client.id == client_fk) & (Client.owner == owner)) is None:
-            return _error_response(f"Client {client_fk} not found.", status_code=404)
-    driver_fk = int(driver_id) if driver_id.strip() else None
+    client_fk, error = _resolve_client(client_id, owner)
+    if error is not None:
+        return error
+    driver_fk, error = _resolve_driver(driver_id)
+    if error is not None:
+        return error
 
     Printer.create(
         name=name,
         ip_address=ip_address,
         port_name=port_name,
         duplex_mode=duplex_mode,
-        color_mode=color_mode_bool,
+        # HTML checkbox convention: "on" = True, absent/empty = False
+        color_mode=color_mode == "on",
         paper_size=paper_size,
-        collate=collate_bool,
+        collate=collate == "on",
         owner=owner,
         client=client_fk,
         driver=driver_fk,
@@ -161,21 +202,16 @@ def update_printer(
     ip_address = ip_address.strip() or printer.ip_address
     port_name = port_name.strip() or printer.port_name
 
-    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}.")
+    invalid = _validate_fields(name, ip_address, port_name, duplex_mode, paper_size)
+    if invalid is not None:
+        return invalid
 
-    client_fk = int(client_id) if client_id.strip() else None
-    if client_fk is not None:
-        if Client.get_or_none((Client.id == client_fk) & (Client.owner == owner)) is None:
-            return _error_response(f"Client {client_fk} not found.", status_code=404)
+    client_fk, error = _resolve_client(client_id, owner)
+    if error is not None:
+        return error
+    driver_fk, error = _resolve_driver(driver_id)
+    if error is not None:
+        return error
 
     printer.name = name
     printer.ip_address = ip_address
@@ -185,7 +221,7 @@ def update_printer(
     printer.paper_size = paper_size
     printer.collate = collate == "on"
     printer.client = client_fk
-    printer.driver = int(driver_id) if driver_id.strip() else None
+    printer.driver = driver_fk
     printer.updated_at = datetime.now(UTC).replace(tzinfo=None)
     printer.save()
 

imptune/api/session.py

@@ -7,9 +7,8 @@ from fastapi import APIRouter, Form, Request
 from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
 from fastapi.templating import Jinja2Templates
 
-import imptune.config as cfg
 from imptune.db.models import Owner
-from imptune.services.session import COOKIE_MAX_AGE, COOKIE_NAME, is_same_origin
+from imptune.services.session import COOKIE_NAME, cookie_kwargs, is_same_origin
 
 router = APIRouter(prefix="/session")
 
@@ -62,12 +61,5 @@ def restore_session(request: Request, key: str = Form(...)):
         )
 
     response = RedirectResponse(url="/", status_code=303)
-    response.set_cookie(
-        COOKIE_NAME,
-        owner.key,
-        max_age=COOKIE_MAX_AGE,
-        httponly=True,
-        samesite="lax",
-        secure=cfg.COOKIE_SECURE,
-    )
+    response.set_cookie(COOKIE_NAME, owner.key, **cookie_kwargs())
     return response