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"
",
+ 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"",
+ 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
+
+
+
+
+ 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 %}
+
+
+
+ | Name |
+ Created |
+
+
+
+ {% for c in clients %}
+
+ | {{ c.name }} |
+ {{ c.created_at.strftime('%Y-%m-%d') if c.created_at else '—' }} |
+
+ {% endfor %}
+
+
+ {% 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 }}
+
+
+
+ | Name |
+ IP Address |
+ Port |
+ Driver |
+ Duplex |
+ Color |
+ Paper |
+ Collate |
+ Actions |
+
+
+
+ {% for p in printers %}
+
+ | {{ 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 %}
+
+
+
+ {% 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}")