Fixing imports

app/services/danger_zone.py

@@ -0,0 +1,57 @@
+"""Irreversible bulk-delete of all domain (parts/taxonomy/fit) data.
+
+Leaves users, roles/permissions, refresh tokens, the audit log, and the
+ImportedPartConfig (OData connection settings) untouched.
+"""
+
+from sqlalchemy import delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.associations import augmented_part_imported_parts
+from app.models.attachment_point import AttachmentPoint
+from app.models.attachment_point_name import AttachmentPointName
+from app.models.augmented_part import AugmentedPart, AugmentedPartNote
+from app.models.clamp_connection import ClampConnection
+from app.models.fit import Fit, FitNote
+from app.models.imported_part import ImportedPart
+from app.models.iso_thread import IsoThread
+from app.models.manufacturer import Manufacturer
+from app.models.part_colour import PartColour
+from app.models.part_type import PartType
+from app.models.scope_info import ScopeInfo
+from app.models.scope_model import ScopeModel
+from app.models.static_connection import StaticConnection
+from app.models.sub_part_type import SubPartType
+from app.models.unc_thread import UncThread
+
+# Children before the RESTRICT-protected reference tables they point to.
+_TARGETS = [
+    ("Fit notes", FitNote),
+    ("Fits", Fit),
+    ("Augmented part notes", AugmentedPartNote),
+    ("Scope info", ScopeInfo),
+    ("Attachment points", AttachmentPoint),
+    ("Augmented part / imported part links", augmented_part_imported_parts),
+    ("Augmented parts", AugmentedPart),
+    ("Imported parts", ImportedPart),
+    ("Scope models", ScopeModel),
+    ("Attachment point names", AttachmentPointName),
+    ("ISO threads", IsoThread),
+    ("UNC threads", UncThread),
+    ("Clamp connections", ClampConnection),
+    ("Static connections", StaticConnection),
+    ("Manufacturers", Manufacturer),
+]
+
+
+async def purge_domain_data(db: AsyncSession) -> dict[str, int]:
+    """Delete every row from every domain table, in FK-safe order.
+
+    Returns a label -> rows-deleted map for reporting/audit purposes.
+    """
+    counts: dict[str, int] = {}
+    for label, target in _TARGETS:
+        result = await db.execute(delete(target))
+        counts[label] = result.rowcount or 0
+    await db.commit()
+    return counts

app/templates/_nav.html

@@ -145,6 +145,9 @@
         {% if su or "audit_log:read" in p %}
           <a href="/admin/audit-log">{{ icon("list-checks") }} Audit Log</a>
         {% endif %}
+        {% if su %}
+          <a href="/admin/danger-zone">{{ icon("alert-circle") }} Danger Zone</a>
+        {% endif %}
       </div>
     </div>
     {% endif %}

app/templates/admin/_danger_zone_result.html

@@ -0,0 +1,14 @@
+{% if total %}
+<p style="color:var(--color-danger,#c00)">
+  Purge complete: {{ total }} row(s) deleted.
+</p>
+<ul class="muted">
+  {% for label, n in counts.items() %}
+  {% if n %}
+  <li>{{ label }}: {{ n }}</li>
+  {% endif %}
+  {% endfor %}
+</ul>
+{% else %}
+<p class="muted">Nothing to purge — the database was already clean.</p>
+{% endif %}

app/templates/admin/danger_zone.html

@@ -0,0 +1,32 @@
+{% extends "base.html" %}
+{% block title %}Danger Zone · {{ app_name }}{% endblock %}
+{% block body %}
+{% include "_nav.html" %}
+<main class="container">
+  <h1>Danger Zone</h1>
+
+  <div class="card" style="border: 1px solid var(--color-danger, #c00)">
+    <h2>Purge database</h2>
+    <p>
+      Permanently deletes every part, taxonomy entry, attachment point, fit,
+      and imported part in the database — manufacturers, part colours, part
+      types, sub part types, scope models, ISO/UNC threads, clamp/static
+      connections, attachment point names, augmented parts, scope info,
+      attachment points, fits, fit notes, and imported parts.
+    </p>
+    <p>
+      <strong>Not</strong> deleted: users, roles &amp; permissions, refresh
+      tokens, the audit log, and the OData connection configuration.
+    </p>
+    <p class="muted">This cannot be undone. Take a database backup first if you might need this data again.</p>
+    <button class="btn btn-danger"
+            hx-post="/admin/danger-zone/purge"
+            hx-target="#purge-result"
+            hx-swap="innerHTML"
+            hx-confirm="This will PERMANENTLY delete all parts, taxonomy, attachment points, fits, and imported parts. Users and the OData configuration are kept. This cannot be undone. Continue?">
+      Purge all part/fit data
+    </button>
+    <div id="purge-result" style="margin-top:1rem"></div>
+  </div>
+</main>
+{% endblock %}

app/web/admin.py

@@ -43,6 +43,7 @@ from app.models.fit import FitLabel
 from app.models.unc_thread import UncSeries
 from app.models.user import User
 from app.services import augmented_part_sync as augmented_part_sync_service
+from app.services import danger_zone as danger_zone_service
 from app.services import fit_matching as fit_matching_service
 from app.services import jobs
 from app.services import odata as odata_service
@@ -131,6 +132,16 @@ def web_require(*perms: str):
     return checker
 
 
+async def web_require_superuser(user: User = Depends(get_current_web_user)) -> User:
+    """Web dependency: superusers only, no permission-string bypass."""
+    if not user.is_superuser:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You do not have access to this page.",
+        )
+    return user
+
+
 async def _role_perm_ctx(db: AsyncSession) -> dict:
     all_perms = await role_crud.list_permissions(db)
     scopes = sorted({p.name.split(":")[0] for p in all_perms})
@@ -3598,3 +3609,45 @@ async def audit_log_page(
         "admin/audit_log.html",
         {"user": user, "entries": await audit_crud.list_recent(db)},
     )
+
+
+# --- Danger Zone ---------------------------------------------------------------
+
+
+@router.get("/danger-zone", response_class=HTMLResponse)
+async def danger_zone_page(
+    request: Request,
+    user: User = Depends(web_require_superuser),
+):
+    return templates.TemplateResponse(
+        request,
+        "admin/danger_zone.html",
+        {"user": user},
+    )
+
+
+@router.post("/danger-zone/purge", response_class=HTMLResponse)
+async def purge_database(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    actor: User = Depends(web_require_superuser),
+):
+    counts = await danger_zone_service.purge_domain_data(db)
+    total = sum(counts.values())
+    breakdown = ", ".join(f"{label} {n}" for label, n in counts.items() if n)
+    await audit_crud.log(
+        db,
+        actor=actor,
+        action="delete",
+        resource_type="Database",
+        resource_id=None,
+        description=(
+            f"Purged all domain data via Danger Zone ({total} row(s) deleted"
+            + (f": {breakdown})" if breakdown else ")")
+        ),
+    )
+    return templates.TemplateResponse(
+        request,
+        "admin/_danger_zone_result.html",
+        {"total": total, "counts": counts},
+    )

tests/test_danger_zone.py

@@ -0,0 +1,157 @@
+import pytest
+
+from app.crud import audit as audit_crud
+from app.models.attachment_point import AttachmentPoint
+from app.models.attachment_point_name import AttachmentPointName
+from app.models.augmented_part import AugmentedPart, AugmentedPartNote
+from app.models.clamp_connection import ClampConnection
+from app.models.fit import Fit, FitLabel, FitNote
+from app.models.imported_part import ImportedPart
+from app.models.imported_part_config import ImportedPartConfig
+from app.models.iso_thread import IsoThread
+from app.models.manufacturer import Manufacturer
+from app.models.part_colour import PartColour
+from app.models.part_type import PartType
+from app.models.scope_info import ScopeInfo
+from app.models.scope_model import ScopeModel
+from app.models.static_connection import StaticConnection
+from app.models.sub_part_type import SubPartType
+from app.models.unc_thread import UncSeries, UncThread
+
+
+async def _seed_domain_data(db):
+    manufacturer = Manufacturer(name="Acme", url="https://acme.test")
+    colour = PartColour(name="Black")
+    scope_t = PartType(name="Scope")
+    cover_t = PartType(name="Flip Cover")
+    apn = AttachmentPointName(name="Objective")
+    db.add_all([manufacturer, colour, scope_t, cover_t, apn])
+    await db.flush()
+
+    sub_t = SubPartType(name="Riflescope", part_type_id=scope_t.id)
+    scope_model = ScopeModel(name="Alpha", manufacturer_id=manufacturer.id)
+    iso = IsoThread(diameter_mm=52.0, pitch_mm=0.75)
+    unc = UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC)
+    clamp = ClampConnection(min_mm=25.0, max_mm=30.0)
+    static = StaticConnection(od_mm=50.0, od_in=1.97)
+    db.add_all([sub_t, scope_model, iso, unc, clamp, static])
+    await db.flush()
+
+    scope = AugmentedPart(
+        description="Alpha 3-18x50",
+        sku="ALP-318",
+        colour_id=colour.id,
+        manufacturer_id=manufacturer.id,
+        part_type_id=scope_t.id,
+        scope_model_id=scope_model.id,
+        sub_part_type_id=sub_t.id,
+    )
+    cover = AugmentedPart(
+        description="Flip cover 50mm",
+        colour_id=colour.id,
+        manufacturer_id=manufacturer.id,
+        part_type_id=cover_t.id,
+    )
+    db.add_all([scope, cover])
+    await db.flush()
+
+    db.add(AugmentedPartNote(augmented_part_id=scope.id, username="tester", body="note"))
+    db.add(ScopeInfo(augmented_part_id=scope.id, magnification_min=3, magnification_max=18))
+
+    ap_s = AttachmentPoint(
+        augmented_part_id=scope.id,
+        attachment_point_name_id=apn.id,
+        iso_thread_id=iso.id,
+    )
+    ap_c = AttachmentPoint(
+        augmented_part_id=cover.id,
+        attachment_point_name_id=apn.id,
+        iso_thread_id=iso.id,
+    )
+    db.add_all([ap_s, ap_c])
+    await db.flush()
+
+    fit = Fit(
+        attachment_point_a_id=ap_s.id,
+        attachment_point_b_id=ap_c.id,
+        augmented_part_a_id=scope.id,
+        augmented_part_b_id=cover.id,
+        label=FitLabel.POSSIBLE,
+    )
+    db.add(fit)
+    await db.flush()
+    db.add(FitNote(fit_id=fit.id, username="tester", body="fits fine"))
+
+    imported = ImportedPart(InventoryID="INV-1", Description="Alpha 3-18x50")
+    db.add(imported)
+
+    config = ImportedPartConfig(id=1, odata_url="https://example.test/odata", username="svc")
+    db.add(config)
+
+    await db.commit()
+
+
+async def _web_login(client, email, password="secret123"):
+    resp = await client.post("/login", data={"email": email, "password": password})
+    assert resp.status_code == 303
+
+
+@pytest.mark.asyncio
+async def test_non_superuser_forbidden(client, seeded, make_user):
+    # "admin" role holds every permission but is not a superuser.
+    await make_user("admin@example.com", role_ids=[seeded["roles"]["admin"].id])
+    await _web_login(client, "admin@example.com")
+
+    assert (await client.get("/admin/danger-zone")).status_code == 403
+    assert (await client.post("/admin/danger-zone/purge")).status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_purge_wipes_domain_data_but_keeps_users_and_odata_config(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        await _seed_domain_data(db)
+
+    await make_user("root@example.com", is_superuser=True)
+    await _web_login(client, "root@example.com")
+
+    assert (await client.get("/admin/danger-zone")).status_code == 200
+
+    resp = await client.post("/admin/danger-zone/purge")
+    assert resp.status_code == 200
+    assert "Purge complete" in resp.text
+
+    async with db_sessionmaker() as db:
+        for Model in (
+            AugmentedPart,
+            AugmentedPartNote,
+            ScopeInfo,
+            AttachmentPoint,
+            Fit,
+            FitNote,
+            ImportedPart,
+            ScopeModel,
+            SubPartType,
+            PartType,
+            AttachmentPointName,
+            IsoThread,
+            UncThread,
+            ClampConnection,
+            StaticConnection,
+            PartColour,
+            Manufacturer,
+        ):
+            rows = (await db.execute(Model.__table__.select())).all()
+            assert rows == [], f"{Model.__name__} should be empty after purge"
+
+        # Kept: users, RBAC, and the OData connection config.
+        from app.crud import user as user_crud
+
+        assert await user_crud.get_by_email(db, "root@example.com") is not None
+        config = await db.get(ImportedPartConfig, 1)
+        assert config is not None
+        assert config.odata_url == "https://example.test/odata"
+
+        entries = await audit_crud.list_recent(db)
+        assert any("Purged all domain data" in e.description for e in entries)