kawaandClaude Opus 5
2c06806814
Driver rename and icons: `Driver.display_name` plus a `DriverIcon` table, both global/shared like the `Driver` row they hang off, so a rename or an icon is what every Owner sees. The rename/icon dialog keeps its forms as siblings (nested forms are invalid HTML) and the icon routes return an `hx-swap-oob` thumbnail refresh rather than re-rendering the table, which would tear the open `<dialog>` out of the DOM. Web image picker: `GET /web/images` renders a pickable grid for a printer or a driver icon, with the search term prefilled from the entity name and editable. Picking one downloads it server-side and normalizes it. Driver download search: `GET /web/drivers` searches for a vendor-wide driver (the term is rewritten into the vendor's real product name for 15 brands) or for the exact model as typed. Links only — nothing is downloaded, and the fragment says the results are unvetted. Icon uploads no longer reject off-size or non-PNG files: `normalize_icon()` letterboxes any decodable raster into a 256x256 PNG. An already-exact 256x256 PNG is returned byte-identical, because icon storage is content-addressed and re-encoding would move the file on every save. `fetch_image()` makes the request from the server, so `assert_fetchable()` refuses any URL resolving to a private, loopback, or link-local address, and re-runs on every redirect. ImpTune sits on the same LAN as the printers it configures; an unguarded fetcher would be a port scanner for anyone who can reach the UI. DuckDuckGo is scraped, not called through an API — no key needed, but fragile, so both search functions swallow parse failures and return [] instead of 500ing a page. `WEB_SEARCH=false` disables every outbound request and hides the controls, for air-gapped installs. Also: one shared `Jinja2Templates` in `templating.py` instead of five per-router instances, so a template global is declared once; `_add_missing_columns()` in `database.py` adds new nullable columns to a pre-existing table, which `create_tables(safe=True)` skips; `db_env` in test_db.py now closes its connection on teardown, or the next test's ORM writes land in the previous test's DB file. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
"""Peewee SQLite database instance and initialization."""
|
|
from peewee import SqliteDatabase
|
|
|
|
from imptune.config import DB_PATH
|
|
|
|
# Deferred init — path is set at runtime via init_db() so tests can override DB_PATH
|
|
db = SqliteDatabase(None)
|
|
|
|
|
|
def init_db() -> None:
|
|
"""Initialize the SQLite database with WAL mode and foreign keys.
|
|
|
|
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, DriverIcon, Owner, 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={
|
|
"journal_mode": "wal",
|
|
"foreign_keys": 1,
|
|
},
|
|
)
|
|
db.connect(reuse_if_open=True)
|
|
db.create_tables([Owner, Client, Driver, Printer, Icon, DriverIcon], safe=True)
|
|
_migrate_owner_column(cfg.DATA_DIR)
|
|
_add_missing_columns()
|
|
|
|
|
|
def _add_missing_columns() -> None:
|
|
"""Add columns introduced after a DB was first created.
|
|
|
|
`create_tables(safe=True)` skips an existing table entirely, so a new field
|
|
on an old model never lands without this. Plain ADD COLUMN of a nullable
|
|
field — no backfill needed.
|
|
"""
|
|
added: dict[str, str] = {"driver": "display_name VARCHAR(255)"}
|
|
for table, column_def in added.items():
|
|
column = column_def.split()[0]
|
|
columns = {row[1] for row in db.execute_sql(f"PRAGMA table_info({table})")}
|
|
if column not in columns:
|
|
db.execute_sql(f"ALTER TABLE {table} ADD COLUMN {column_def}")
|
|
|
|
|
|
def _migrate_owner_column(data_dir: str) -> None:
|
|
"""Add owner_id to printer/client if missing, backfilling pre-existing rows.
|
|
|
|
Runs against a DB created before per-owner scoping existed. Idempotent:
|
|
a no-op once the column exists and no rows are left with a NULL owner_id.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
from imptune.db.models import Client, Owner, Printer
|
|
from imptune.services.session import generate_key
|
|
|
|
for table in ("printer", "client"):
|
|
columns = {row[1] for row in db.execute_sql(f"PRAGMA table_info({table})")}
|
|
if "owner_id" not in columns:
|
|
db.execute_sql(
|
|
f"ALTER TABLE {table} ADD COLUMN owner_id INTEGER REFERENCES owner (id)"
|
|
)
|
|
|
|
orphaned = Printer.select().where(Printer.owner.is_null()).count() or Client.select().where(
|
|
Client.owner.is_null()
|
|
).count()
|
|
if not orphaned:
|
|
return
|
|
|
|
legacy_owner = Owner.create(key=generate_key(), is_permanent=True)
|
|
Printer.update(owner=legacy_owner.id).where(Printer.owner.is_null()).execute()
|
|
Client.update(owner=legacy_owner.id).where(Client.owner.is_null()).execute()
|
|
|
|
key_path = Path(data_dir) / "legacy_owner_key.txt"
|
|
key_path.write_text(legacy_owner.key, encoding="utf-8")
|
|
print(f"[imptune] Pre-existing printers/clients migrated to a legacy owner. "
|
|
f"Restore key written to {key_path} — paste it into /session/restore to reclaim them.")
|