Adding archived flag to augmented parts, fixing migrations so that creation from scratch is successful

alembic/versions/0c3d1a5cff1f_add_part_colours.py

@@ -18,12 +18,14 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "part_colours",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("name"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("part_colours")
-    pass
-    # ### end Alembic commands ###

alembic/versions/5b6a7817b408_add_static_connections.py

@@ -18,12 +18,15 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "static_connections",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("od_mm", sa.Float(), nullable=False),
+        sa.Column("od_in", sa.Float(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("od_mm"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("static_connections")
-    pass
-    # ### end Alembic commands ###

alembic/versions/70e48bda8829_add_attachment_point_names.py

@@ -18,12 +18,14 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "attachment_point_names",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("name"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("attachment_point_names")
-    pass
-    # ### end Alembic commands ###

alembic/versions/7be5ef599d3f_add_imported_parts_tables.py

@@ -18,12 +18,28 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "imported_parts",
-    # ### end Alembic commands ###
+        sa.Column("InventoryID", sa.String(), nullable=False),
+        sa.Column("Description", sa.String(), nullable=True),
+        sa.Column("TBXColour", sa.String(), nullable=True),
+        sa.Column("ProductMaxThreadDepthmm", sa.String(), nullable=True),
+        sa.Column("ProductMaxThreadDepthin", sa.String(), nullable=True),
+        sa.Column("ProductODmm", sa.String(), nullable=True),
+        sa.Column("ProductODin", sa.String(), nullable=True),
+        sa.Column("TBXThreadAttribute", sa.String(), nullable=True),
+        sa.PrimaryKeyConstraint("InventoryID"),
+    )
+    op.create_table(
+        "imported_part_config",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("odata_url", sa.String(), nullable=True),
+        sa.Column("username", sa.String(), nullable=True),
+        sa.Column("password", sa.String(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("imported_part_config")
-    pass
+    op.drop_table("imported_parts")
-    # ### end Alembic commands ###

alembic/versions/8bc411b5814e_add_augmented_parts.py

@@ -18,12 +18,34 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "augmented_parts",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("sku", sa.String(), nullable=True),
+        sa.Column("description", sa.String(), nullable=False),
+        sa.Column("colour_id", sa.Integer(), nullable=False),
+        sa.Column("manufacturer_id", sa.Integer(), nullable=False),
+        sa.Column("scope_model_id", sa.Integer(), nullable=True),
+        sa.Column("part_type_id", sa.Integer(), nullable=False),
+        sa.Column("sub_part_type_id", sa.Integer(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["colour_id"], ["part_colours.id"], ondelete="RESTRICT"
+        ),
+        sa.ForeignKeyConstraint(
+            ["manufacturer_id"], ["manufacturers.id"], ondelete="RESTRICT"
+        ),
+        sa.ForeignKeyConstraint(
+            ["scope_model_id"], ["scope_models.id"], ondelete="SET NULL"
+        ),
+        sa.ForeignKeyConstraint(
+            ["part_type_id"], ["part_types.id"], ondelete="RESTRICT"
+        ),
+        sa.ForeignKeyConstraint(
+            ["sub_part_type_id"], ["sub_part_types.id"], ondelete="SET NULL"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("augmented_parts")
-    pass
-    # ### end Alembic commands ###

alembic/versions/9a1c2f6e0b3d_add_archived_to_augmented_parts.py

@@ -0,0 +1,32 @@
+"""add_archived_to_augmented_parts
+
+Revision ID: 9a1c2f6e0b3d
+Revises: 235d823a56ee
+Create Date: 2026-08-17 21:30:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "9a1c2f6e0b3d"
+down_revision: str | None = "235d823a56ee"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("augmented_parts", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "archived", sa.Boolean(), nullable=False, server_default=sa.false()
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("augmented_parts", schema=None) as batch_op:
+        batch_op.drop_column("archived")

alembic/versions/adf4a5b9493e_add_attachment_points.py

@@ -18,12 +18,38 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "attachment_points",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("augmented_part_id", sa.Integer(), nullable=False),
+        sa.Column("attachment_point_name_id", sa.Integer(), nullable=False),
+        sa.Column("iso_thread_id", sa.Integer(), nullable=True),
+        sa.Column("unc_thread_id", sa.Integer(), nullable=True),
+        sa.Column("clamp_connection_id", sa.Integer(), nullable=True),
+        sa.Column("static_connection_id", sa.Integer(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["augmented_part_id"], ["augmented_parts.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["attachment_point_name_id"],
+            ["attachment_point_names.id"],
+            ondelete="RESTRICT",
+        ),
+        sa.ForeignKeyConstraint(
+            ["iso_thread_id"], ["iso_threads.id"], ondelete="RESTRICT"
+        ),
+        sa.ForeignKeyConstraint(
+            ["unc_thread_id"], ["unc_threads.id"], ondelete="RESTRICT"
+        ),
+        sa.ForeignKeyConstraint(
+            ["clamp_connection_id"], ["clamp_connections.id"], ondelete="RESTRICT"
+        ),
+        sa.ForeignKeyConstraint(
+            ["static_connection_id"], ["static_connections.id"], ondelete="RESTRICT"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("attachment_points")
-    pass
-    # ### end Alembic commands ###

alembic/versions/b60d0aa3ceda_add_sub_part_types_table.py

@@ -18,12 +18,18 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "sub_part_types",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("part_type_id", sa.Integer(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["part_type_id"], ["part_types.id"], ondelete="RESTRICT"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("name", "part_type_id"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("sub_part_types")
-    pass
-    # ### end Alembic commands ###

alembic/versions/bfbce324cd5f_add_augmented_part_imported_parts.py

@@ -18,12 +18,19 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "augmented_part_imported_parts",
-    # ### end Alembic commands ###
+        sa.Column("augmented_part_id", sa.Integer(), nullable=False),
+        sa.Column("imported_part_id", sa.String(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["augmented_part_id"], ["augmented_parts.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["imported_part_id"], ["imported_parts.InventoryID"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("augmented_part_id", "imported_part_id"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("augmented_part_imported_parts")
-    pass
-    # ### end Alembic commands ###

alembic/versions/c7882cedc9ae_add_clamp_connections.py

@@ -18,12 +18,16 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "clamp_connections",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("min_mm", sa.Float(), nullable=True),
+        sa.Column("max_mm", sa.Float(), nullable=True),
+        sa.Column("min_in", sa.Float(), nullable=True),
+        sa.Column("max_in", sa.Float(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("clamp_connections")
-    pass
-    # ### end Alembic commands ###

alembic/versions/fd5379986da4_add_audit_logs_table.py

@@ -18,12 +18,29 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
-    pass
+        "audit_logs",
-    # ### end Alembic commands ###
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("user_email", sa.String(length=255), nullable=False),
+        sa.Column("action", sa.String(length=16), nullable=False),
+        sa.Column("resource_type", sa.String(length=64), nullable=False),
+        sa.Column("resource_id", sa.Integer(), nullable=True),
+        sa.Column("description", sa.String(), nullable=False),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_audit_logs_created_at", "audit_logs", ["created_at"])
+    op.create_index("ix_audit_logs_user_id", "audit_logs", ["user_id"])
 
 
 def downgrade() -> None:
-    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index("ix_audit_logs_user_id", table_name="audit_logs")
-    pass
+    op.drop_index("ix_audit_logs_created_at", table_name="audit_logs")
-    # ### end Alembic commands ###
+    op.drop_table("audit_logs")

app/api/augmented_parts.py

@@ -24,10 +24,13 @@ router = APIRouter()
 async def list_augmented_parts(
     skip: int = 0,
     limit: int = 100,
+    include_archived: bool = False,
     db: AsyncSession = Depends(get_db),
     _: User = Depends(require_permission("augmented_parts:read")),
 ):
-    return await augmented_part_crud.list_all(db, skip=skip, limit=limit)
+    return await augmented_part_crud.list_all(
+        db, skip=skip, limit=limit, include_archived=include_archived
+    )
 
 
 @router.post("", response_model=AugmentedPartRead, status_code=status.HTTP_201_CREATED)
@@ -72,7 +75,7 @@ async def sync_augmented_parts(
         resource_type="AugmentedPart",
         resource_id=None,
         description=(
-            f"Sync: {result.created} created, {result.deleted} deleted, "
+            f"Sync: {result.created} created, {result.archived} archived, "
             f"{result.skipped} skipped (no type match)"
         ),
     )

app/api/imported_parts.py

@@ -13,6 +13,7 @@ from app.schemas.imported_part import (
     ImportedPartRead,
     SyncResult,
 )
+from app.services import augmented_part_sync
 from app.services import odata as odata_service
 
 router = APIRouter()
@@ -128,7 +129,9 @@ async def delete_imported_part(
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail="Imported part not found"
         )
-    await imported_part_crud.delete(db, part)
+    archived = await augmented_part_sync.delete_imported_part_and_archive_orphans(
+        db, part
+    )
     await audit_crud.log(
         db,
         actor=actor,
@@ -137,3 +140,15 @@ async def delete_imported_part(
         resource_id=None,
         description=audit_crud.describe_delete("ImportedPart", 0, inventory_id),
     )
+    if archived:
+        await audit_crud.log(
+            db,
+            actor=actor,
+            action="update",
+            resource_type="AugmentedPart",
+            resource_id=None,
+            description=(
+                f"Archived {archived} augmented part(s) left with no remaining "
+                f"imported parts after deleting {inventory_id!r}"
+            ),
+        )

app/crud/augmented_part.py

@@ -58,14 +58,15 @@ async def get_by_sku(db: AsyncSession, sku: str) -> AugmentedPart | None:
 
 
 async def list_all(
-    db: AsyncSession, skip: int = 0, limit: int = 100
+    db: AsyncSession,
+    skip: int = 0,
+    limit: int = 100,
+    include_archived: bool = False,
 ) -> list[AugmentedPart]:
-    result = await db.execute(
+    q = select(AugmentedPart).order_by(AugmentedPart.description)
-        select(AugmentedPart)
+    if not include_archived:
-        .order_by(AugmentedPart.description)
+        q = q.where(AugmentedPart.archived.is_(False))
-        .offset(skip)
+    result = await db.execute(q.offset(skip).limit(limit))
-        .limit(limit)
-    )
     return list(result.scalars().all())
 
 
@@ -85,6 +86,7 @@ async def list_filtered(
     scope_model_id: int | None = None,
     part_type_id: int | None = None,
     sub_part_type_id: int | None = None,
+    include_archived: bool = False,
     sort: str = "description",
     direction: str = "asc",
     skip: int = 0,
@@ -106,6 +108,8 @@ async def list_filtered(
         q = q.where(AugmentedPart.part_type_id == part_type_id)
     if sub_part_type_id is not None:
         q = q.where(AugmentedPart.sub_part_type_id == sub_part_type_id)
+    if not include_archived:
+        q = q.where(AugmentedPart.archived.is_(False))
     q = _apply_sort(q, sort, direction)
     result = await db.execute(q.offset(skip).limit(limit))
     return list(result.scalars().all())
@@ -119,6 +123,7 @@ async def count_filtered(
     scope_model_id: int | None = None,
     part_type_id: int | None = None,
     sub_part_type_id: int | None = None,
+    include_archived: bool = False,
 ) -> int:
     from sqlalchemy import func
 
@@ -138,6 +143,8 @@ async def count_filtered(
         q = q.where(AugmentedPart.part_type_id == part_type_id)
     if sub_part_type_id is not None:
         q = q.where(AugmentedPart.sub_part_type_id == sub_part_type_id)
+    if not include_archived:
+        q = q.where(AugmentedPart.archived.is_(False))
     result = await db.execute(q)
     return result.scalar_one()
 
@@ -251,6 +258,15 @@ async def get_unlinked_imported_parts(db: AsyncSession) -> list[ImportedPart]:
     return list(result.scalars().all())
 
 
+async def set_archived(
+    db: AsyncSession, ap: AugmentedPart, archived: bool
+) -> AugmentedPart:
+    ap.archived = archived
+    await db.commit()
+    await db.refresh(ap, attribute_names=_RELATIONSHIPS)
+    return ap
+
+
 async def delete(db: AsyncSession, ap: AugmentedPart) -> None:
     await db.delete(ap)
     await db.commit()

app/crud/imported_part.py

@@ -105,11 +105,6 @@ async def count_filtered(db: AsyncSession, description: str = "") -> int:
     return result.scalar_one()
 
 
-async def delete(db: AsyncSession, part: ImportedPart) -> None:
-    await db.delete(part)
-    await db.commit()
-
-
 async def upsert_bulk(db: AsyncSession, records: list[dict]) -> tuple[int, int]:
     """Insert-or-update records keyed on InventoryID. Returns (created, updated)."""
     if not records:

app/models/augmented_part.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, String, Text, func
+from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.core.database import Base
@@ -36,6 +36,10 @@ class AugmentedPart(Base):
         nullable=True,
         default=None,
     )
+    # Set instead of deleting the row when the last backing ImportedPart is
+    # removed, so fit/attachment-point history is preserved. Hidden from list
+    # views by default.
+    archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
 
     # linked imported parts — not eagerly loaded; used by sync service only
     imported_parts: Mapped[list[ImportedPart]] = relationship(

app/schemas/augmented_part.py

@@ -53,11 +53,12 @@ class AugmentedPartUpdate(BaseModel):
     scope_model_id: int | None = None
     part_type_id: int | None = None
     sub_part_type_id: int | None = None
+    archived: bool | None = None
 
 
 class AugmentedPartSyncResult(BaseModel):
     created: int
-    deleted: int
+    archived: int
     skipped: int
 
 
@@ -65,6 +66,7 @@ class AugmentedPartRead(AugmentedPartBase):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
+    archived: bool
     colour: PartColourRead
     manufacturer: ManufacturerRead
     scope_model: ScopeModelRead | None

app/services/augmented_part_sync.py

@@ -2,11 +2,12 @@
 
 Algorithm
 ---------
-Phase 1 – prune stale entries
+Phase 1 – archive stale entries
   • Find association rows whose imported_part_id no longer exists.
   • Delete those rows.
-  • Delete any AugmentedPart that ends up with zero remaining associations
+  • Archive (rather than delete) any AugmentedPart that ends up with zero
-    (i.e. it was created purely from now-deleted ImportedParts).
+    remaining associations (i.e. it was created purely from now-deleted
+    ImportedParts), preserving its fit/attachment-point history.
 
 Phase 2 – create entries for new ImportedParts
   • Any ImportedPart not yet linked to an AugmentedPart gets one created.
@@ -33,12 +34,12 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.crud import attachment_point as attachment_point_crud
 from app.crud import attachment_point_name as attachment_point_name_crud
+from app.crud import clamp_connection as clamp_connection_crud
 from app.crud import iso_thread as iso_thread_crud
 from app.crud import manufacturer as manufacturer_crud
 from app.crud import part_colour as part_colour_crud
 from app.crud import part_type as part_type_crud
 from app.crud import static_connection as static_connection_crud
-from app.crud import clamp_connection as clamp_connection_crud
 from app.crud import unc_thread as unc_thread_crud
 from app.models.associations import augmented_part_imported_parts
 from app.models.augmented_part import AugmentedPart
@@ -268,6 +269,91 @@ def _infer_part_type(description: str, part_types: list[PartType]) -> PartType |
     return None
 
 
+async def _archive_orphans(db: AsyncSession, candidate_ap_ids: set[int]) -> int:
+    """Archive any of `candidate_ap_ids` left with zero linked ImportedParts.
+
+    Idempotent: already-archived parts and parts that still have a link are
+    left untouched. Does not commit — the caller commits as part of its own
+    transaction.
+    """
+    archived = 0
+    for ap_id in candidate_ap_ids:
+        result = await db.execute(
+            select(func.count())
+            .select_from(augmented_part_imported_parts)
+            .where(augmented_part_imported_parts.c.augmented_part_id == ap_id)
+        )
+        if result.scalar() == 0:
+            ap = await db.get(AugmentedPart, ap_id)
+            if ap is not None and not ap.archived:
+                ap.archived = True
+                archived += 1
+    return archived
+
+
+async def _detach_and_archive_orphans(
+    db: AsyncSession, imported_part_ids: set[str]
+) -> int:
+    """Detach `imported_part_ids` from any AugmentedPart association and archive
+    AugmentedParts left with zero remaining ImportedPart links as a result.
+
+    Deletes the association rows directly rather than relying on the
+    ImportedPart row's ON DELETE CASCADE, since SQLite (this app's dev/test
+    database) doesn't enforce FK actions by default. Safe to call whether or
+    not the ImportedPart row(s) still exist. Does not commit — the caller
+    commits as part of its own transaction.
+    """
+    if not imported_part_ids:
+        return 0
+
+    result = await db.execute(
+        select(augmented_part_imported_parts.c.augmented_part_id).where(
+            augmented_part_imported_parts.c.imported_part_id.in_(imported_part_ids)
+        )
+    )
+    candidate_ap_ids: set[int] = {row[0] for row in result.all()}
+
+    await db.execute(
+        augmented_part_imported_parts.delete().where(
+            augmented_part_imported_parts.c.imported_part_id.in_(imported_part_ids)
+        )
+    )
+    await db.flush()
+
+    return await _archive_orphans(db, candidate_ap_ids)
+
+
+async def delete_imported_part_and_archive_orphans(
+    db: AsyncSession, imported_part: ImportedPart
+) -> int:
+    """Delete `imported_part` and archive any AugmentedPart left with no
+    remaining ImportedPart links as a result.
+
+    Captures affected AugmentedPart ids before deleting so this is correct
+    regardless of whether the DB enforces the association table's ON DELETE
+    CASCADE. Returns the number of AugmentedParts newly archived.
+    """
+    inventory_id = imported_part.InventoryID
+    result = await db.execute(
+        select(augmented_part_imported_parts.c.augmented_part_id).where(
+            augmented_part_imported_parts.c.imported_part_id == inventory_id
+        )
+    )
+    candidate_ap_ids: set[int] = {row[0] for row in result.all()}
+
+    await db.delete(imported_part)
+    await db.execute(
+        augmented_part_imported_parts.delete().where(
+            augmented_part_imported_parts.c.imported_part_id == inventory_id
+        )
+    )
+    await db.flush()
+
+    archived = await _archive_orphans(db, candidate_ap_ids)
+    await db.commit()
+    return archived
+
+
 async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
     # ------------------------------------------------------------------ phase 1
 
@@ -282,42 +368,10 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
     current_ids: set[str] = {row[0] for row in result.all()}
 
     orphaned_import_ids = all_linked_ids - current_ids
-    deleted = 0
+    archived = 0
 
     if orphaned_import_ids:
-        # Which AugmentedParts are affected?
+        archived = await _detach_and_archive_orphans(db, orphaned_import_ids)
-        result = await db.execute(
-            select(augmented_part_imported_parts.c.augmented_part_id).where(
-                augmented_part_imported_parts.c.imported_part_id.in_(
-                    orphaned_import_ids
-                )
-            )
-        )
-        orphaned_ap_ids: set[int] = {row[0] for row in result.all()}
-
-        # Remove the stale association rows
-        await db.execute(
-            augmented_part_imported_parts.delete().where(
-                augmented_part_imported_parts.c.imported_part_id.in_(
-                    orphaned_import_ids
-                )
-            )
-        )
-        await db.flush()
-
-        # Delete AugmentedParts that now have no remaining associations
-        for ap_id in orphaned_ap_ids:
-            result = await db.execute(
-                select(func.count())
-                .select_from(augmented_part_imported_parts)
-                .where(augmented_part_imported_parts.c.augmented_part_id == ap_id)
-            )
-            if result.scalar() == 0:
-                ap = await db.get(AugmentedPart, ap_id)
-                if ap is not None:
-                    await db.delete(ap)
-                    deleted += 1
-
         await db.commit()
 
     # ------------------------------------------------------------------ phase 2
@@ -334,7 +388,7 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
 
     if not unlinked_ids:
         return AugmentedPartSyncResult(
-            created=created, deleted=deleted, skipped=skipped
+            created=created, archived=archived, skipped=skipped
         )
 
     result = await db.execute(
@@ -385,4 +439,4 @@ async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
 
     await db.commit()
 
-    return AugmentedPartSyncResult(created=created, deleted=deleted, skipped=skipped)
+    return AugmentedPartSyncResult(created=created, archived=archived, skipped=skipped)

app/services/scope_guide.py

@@ -113,6 +113,8 @@ async def build(
         other = other_parts.get(other_id)
         if other is None or other.part_type_id == scope_type.id:
             continue  # scope-to-scope fits have no column
+        if other.archived:
+            continue  # archived parts don't show up as compatible accessories
         by_type = cell_map[scope_id].setdefault(other.part_type_id, {})
         existing = by_type.get(other.id)
         label = fit.label.value

app/templates/admin/_augmented_part_row.html

@@ -7,11 +7,24 @@
   <td>{{ ap.scope_model.name if ap.scope_model else '—' }}</td>
   <td>{{ ap.part_type.name }}</td>
   <td>{{ ap.sub_part_type.name if ap.sub_part_type else '—' }}</td>
+  <td>{% if ap.archived %}<span class="tag">Archived</span>{% else %}—{% endif %}</td>
   <td class="row-actions">
     <button class="btn btn-ghost"
             hx-get="/admin/augmented-parts/{{ ap.id }}/edit"
             hx-target="#augmented-part-{{ ap.id }}"
             hx-swap="outerHTML">Edit</button>
+    {% if ap.archived %}
+    <button class="btn btn-ghost"
+            hx-post="/admin/augmented-parts/{{ ap.id }}/unarchive"
+            hx-target="#augmented-part-{{ ap.id }}"
+            hx-swap="outerHTML">Unarchive</button>
+    {% else %}
+    <button class="btn btn-ghost"
+            hx-post="/admin/augmented-parts/{{ ap.id }}/archive"
+            hx-target="#augmented-part-{{ ap.id }}"
+            hx-swap="outerHTML"
+            hx-confirm="Archive {{ ap.sku or ap.description }}?">Archive</button>
+    {% endif %}
     <button class="btn btn-danger"
             hx-delete="/admin/augmented-parts/{{ ap.id }}"
             hx-target="#augmented-part-{{ ap.id }}"

app/templates/admin/_augmented_part_row_edit.html

@@ -1,5 +1,5 @@
 <tr id="augmented-part-{{ ap.id }}">
-  <td colspan="9">
+  <td colspan="10">
     <form id="augmented-part-edit-form-{{ ap.id }}"
           hx-post="/admin/augmented-parts/{{ ap.id }}"
           hx-target="#augmented-part-{{ ap.id }}"

app/templates/admin/_augmented_parts_table.html

@@ -43,6 +43,7 @@
         <option value="{{ spt.id }}" {{ "selected" if filter_sub_part_type_id and filter_sub_part_type_id|int == spt.id }}>{{ spt.name }}</option>
       {% endfor %}
     </select>
+    <label><input type="checkbox" name="include_archived" {{ "checked" if include_archived }}> Show archived</label>
   </form>
 </div>
 

app/templates/admin/_augmented_parts_table_card.html

@@ -12,6 +12,7 @@
         {{ sort_th("scope_model", "Model", sort_headers) }}
         {{ sort_th("part_type", "Type", sort_headers) }}
         {{ sort_th("sub_part_type", "Sub Type", sort_headers) }}
+        <th>Archived</th>
         <th></th>
       </tr>
     </thead>
@@ -19,7 +20,7 @@
       {% for ap in augmented_parts %}
         {% include "admin/_augmented_part_row.html" %}
       {% else %}
-        <tr><td colspan="9" class="muted">No augmented parts recorded.</td></tr>
+        <tr><td colspan="10" class="muted">No augmented parts recorded.</td></tr>
       {% endfor %}
     </tbody>
   </table>

app/templates/admin/augmented_part_detail.html

@@ -10,6 +10,7 @@
   <p class="muted">
     <span class="tag">{{ ap.part_type.name }}</span>
     {% if ap.sub_part_type %}<span class="tag">{{ ap.sub_part_type.name }}</span>{% endif %}
+    {% if ap.archived %}<span class="tag">Archived</span>{% endif %}
     {% if ap.description %}{{ ap.description }}{% endif %}
   </p>
 

app/templates/admin/augmented_parts.html

@@ -56,7 +56,7 @@
 
   <div class="card">
     <h2>Sync with Imported Parts</h2>
-    <p class="muted">Creates Augmented Part entries for new Imported Parts, and removes entries for deleted Imported Parts.</p>
+    <p class="muted">Creates Augmented Part entries for new Imported Parts, and archives entries for deleted Imported Parts.</p>
     <div class="row-actions">
       <form hx-post="/admin/augmented-parts/sync" hx-target="#augmented-part-list" hx-swap="outerHTML">
         <button type="submit" class="btn btn-primary">Sync now</button>

app/web/admin.py

@@ -1660,6 +1660,7 @@ async def _default_augmented_parts_table_ctx(db: AsyncSession) -> dict:
         ),
         "sort": "description",
         "dir": "asc",
+        "include_archived": False,
     }
 
 
@@ -1668,17 +1669,24 @@ async def augmented_parts_page(
     request: Request,
     sort: str = "description",
     dir: str = "asc",
+    include_archived: bool = False,
     db: AsyncSession = Depends(get_db),
     user: User = Depends(web_require("augmented_parts:read")),
 ):
     ctx = await _augmented_part_form_ctx(db)
-    total = await augmented_part_crud.count_filtered(db)
+    total = await augmented_part_crud.count_filtered(
+        db, include_archived=include_archived
+    )
     ctx["augmented_parts"] = await augmented_part_crud.list_filtered(
-        db, sort=sort, direction=dir, limit=_PAGE_SIZE
+        db,
+        sort=sort,
+        direction=dir,
+        include_archived=include_archived,
+        limit=_PAGE_SIZE,
     )
     ctx["pagination"] = _paginate(
         "/admin/augmented-parts/filter",
-        {"sort": sort, "dir": dir},
+        {"sort": sort, "dir": dir, "include_archived": include_archived},
         1,
         _PAGE_SIZE,
         total,
@@ -1686,7 +1694,7 @@ async def augmented_parts_page(
     )
     ctx["sort_headers"] = _sort_headers(
         "/admin/augmented-parts/filter",
-        {},
+        {"include_archived": include_archived},
         _AUGMENTED_PARTS_SORT_COLUMNS,
         sort,
         dir,
@@ -1694,6 +1702,7 @@ async def augmented_parts_page(
     )
     ctx["sort"] = sort
     ctx["dir"] = dir
+    ctx["include_archived"] = include_archived
     ctx["user"] = user
     return templates.TemplateResponse(request, "admin/augmented_parts.html", ctx)
 
@@ -1764,7 +1773,7 @@ async def sync_augmented_parts(
                 resource_type="AugmentedPart",
                 resource_id=None,
                 description=(
-                    f"Sync: {result.created} created, {result.deleted} deleted, "
+                    f"Sync: {result.created} created, {result.archived} archived, "
                     f"{result.skipped} skipped (no type match)"
                 ),
             )
@@ -1772,7 +1781,7 @@ async def sync_augmented_parts(
                 job_id,
                 message=(
                     f"Sync complete: {result.created} created, "
-                    f"{result.deleted} deleted"
+                    f"{result.archived} archived"
                     + (
                         f", {result.skipped} skipped (type not recognised)"
                         if result.skipped
@@ -1846,6 +1855,7 @@ async def filter_augmented_parts(
     scope_model_id: str = "",
     part_type_id: str = "",
     sub_part_type_id: str = "",
+    include_archived: bool = False,
     sort: str = "description",
     dir: str = "asc",
     page: int = 1,
@@ -1866,6 +1876,7 @@ async def filter_augmented_parts(
         scope_model_id=scope_model_id_int,
         part_type_id=part_type_id_int,
         sub_part_type_id=sub_part_type_id_int,
+        include_archived=include_archived,
     )
     filter_params = {
         "description": description,
@@ -1874,6 +1885,7 @@ async def filter_augmented_parts(
         "scope_model_id": scope_model_id,
         "part_type_id": part_type_id,
         "sub_part_type_id": sub_part_type_id,
+        "include_archived": include_archived,
     }
     pagination = _paginate(
         "/admin/augmented-parts/filter",
@@ -1891,6 +1903,7 @@ async def filter_augmented_parts(
         scope_model_id=scope_model_id_int,
         part_type_id=part_type_id_int,
         sub_part_type_id=sub_part_type_id_int,
+        include_archived=include_archived,
         sort=sort,
         direction=dir,
         skip=pagination["skip"],
@@ -1913,6 +1926,7 @@ async def filter_augmented_parts(
             "sort_headers": sort_headers,
             "sort": sort,
             "dir": dir,
+            "include_archived": include_archived,
         },
     )
 
@@ -2210,6 +2224,60 @@ async def delete_augmented_part(
     return Response(status_code=status.HTTP_200_OK)
 
 
+async def _set_augmented_part_archived(
+    request: Request,
+    augmented_part_id: int,
+    archived: bool,
+    db: AsyncSession,
+    actor: User,
+):
+    ap = await augmented_part_crud.get(db, augmented_part_id)
+    if ap is None:
+        raise HTTPException(status_code=404, detail="Augmented part not found")
+    ap = await augmented_part_crud.set_archived(db, ap, archived)
+    await audit_crud.log(
+        db,
+        actor=actor,
+        action="update",
+        resource_type="AugmentedPart",
+        resource_id=augmented_part_id,
+        description=audit_crud.describe_update(
+            "AugmentedPart", augmented_part_id, {"archived": archived}
+        ),
+    )
+    return templates.TemplateResponse(
+        request, "admin/_augmented_part_row.html", {"ap": ap}
+    )
+
+
+@router.post(
+    "/augmented-parts/{augmented_part_id}/archive", response_class=HTMLResponse
+)
+async def archive_augmented_part(
+    request: Request,
+    augmented_part_id: int,
+    db: AsyncSession = Depends(get_db),
+    actor: User = Depends(web_require("augmented_parts:write")),
+):
+    return await _set_augmented_part_archived(
+        request, augmented_part_id, True, db, actor
+    )
+
+
+@router.post(
+    "/augmented-parts/{augmented_part_id}/unarchive", response_class=HTMLResponse
+)
+async def unarchive_augmented_part(
+    request: Request,
+    augmented_part_id: int,
+    db: AsyncSession = Depends(get_db),
+    actor: User = Depends(web_require("augmented_parts:write")),
+):
+    return await _set_augmented_part_archived(
+        request, augmented_part_id, False, db, actor
+    )
+
+
 def _can_write_attachment_points(user: User) -> bool:
     return user.is_superuser or "attachment_points:write" in user.permissions
 

tests/test_augmented_part_archiving.py

@@ -0,0 +1,257 @@
+import pytest
+
+from app.crud import augmented_part as augmented_part_crud
+from app.models.augmented_part import AugmentedPart
+from app.models.imported_part import ImportedPart
+from app.models.manufacturer import Manufacturer
+from app.models.part_colour import PartColour
+from app.models.part_type import PartType
+from app.services import augmented_part_sync
+
+
+async def _seed_reference_data(db):
+    manufacturer = Manufacturer(name="Acme", url="https://acme.test")
+    colour = PartColour(name="Black")
+    part_type = PartType(name="Scope")
+    db.add_all([manufacturer, colour, part_type])
+    await db.flush()
+    return manufacturer, colour, part_type
+
+
+async def _token(client, email, password="secret123"):
+    resp = await client.post(
+        "/api/auth/login", data={"username": email, "password": password}
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["access_token"]
+
+
+@pytest.mark.asyncio
+async def test_list_and_count_exclude_archived_by_default(db_sessionmaker):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        active = AugmentedPart(
+            description="Active part",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+        )
+        archived = AugmentedPart(
+            description="Archived part",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+            archived=True,
+        )
+        db.add_all([active, archived])
+        await db.commit()
+
+        assert [p.description for p in await augmented_part_crud.list_all(db)] == [
+            "Active part"
+        ]
+        assert await augmented_part_crud.count_filtered(db) == 1
+
+        all_descs = {
+            p.description
+            for p in await augmented_part_crud.list_all(db, include_archived=True)
+        }
+        assert all_descs == {"Active part", "Archived part"}
+        assert await augmented_part_crud.count_filtered(db, include_archived=True) == 2
+
+
+@pytest.mark.asyncio
+async def test_set_archived_toggles_flag(db_sessionmaker):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        ap = AugmentedPart(
+            description="Widget",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+        )
+        db.add(ap)
+        await db.commit()
+
+        ap = await augmented_part_crud.set_archived(db, ap, True)
+        assert ap.archived is True
+
+        ap = await augmented_part_crud.set_archived(db, ap, False)
+        assert ap.archived is False
+
+
+@pytest.mark.asyncio
+async def test_sync_archives_part_only_when_all_backing_imports_are_gone(
+    db_sessionmaker,
+):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        ip_x = ImportedPart(InventoryID="INV-X", Description="Widget X")
+        ip_y = ImportedPart(InventoryID="INV-Y", Description="Widget Y")
+        db.add_all([ip_x, ip_y])
+        await db.flush()
+
+        # One AugmentedPart backed by two ImportedParts.
+        merged = AugmentedPart(
+            description="Merged widget",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+            imported_parts=[ip_x, ip_y],
+        )
+        db.add(merged)
+        await db.commit()
+        merged_id = merged.id
+
+        # Delete one backing ImportedPart directly (simulating an external feed
+        # change) and sync: the AugmentedPart still has INV-Y, so it survives.
+        await db.delete(ip_x)
+        await db.commit()
+
+        result = await augmented_part_sync.sync(db)
+        assert result.archived == 0
+
+        refreshed = await augmented_part_crud.get(db, merged_id)
+        assert refreshed.archived is False
+
+        # Now delete the last remaining backing ImportedPart.
+        await db.delete(ip_y)
+        await db.commit()
+
+        result = await augmented_part_sync.sync(db)
+        assert result.archived == 1
+        assert result.created == 0
+
+        refreshed = await augmented_part_crud.get(db, merged_id)
+        assert refreshed.archived is True
+
+        # Re-running sync is a no-op — it must not double count or explode.
+        result = await augmented_part_sync.sync(db)
+        assert result.archived == 0
+
+
+@pytest.mark.asyncio
+async def test_delete_imported_part_via_api_archives_orphaned_augmented_part(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        ip = ImportedPart(InventoryID="INV-SOLE", Description="Solo widget")
+        db.add(ip)
+        await db.flush()
+        ap = AugmentedPart(
+            description="Solo widget",
+            sku="INV-SOLE",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+            imported_parts=[ip],
+        )
+        db.add(ap)
+        await db.commit()
+        ap_id = ap.id
+
+    await make_user("root@example.com", is_superuser=True)
+    headers = {"Authorization": f"Bearer {await _token(client, 'root@example.com')}"}
+
+    resp = await client.delete("/api/imported-parts/INV-SOLE", headers=headers)
+    assert resp.status_code == 204
+
+    # ImportedPart is really gone.
+    assert (
+        await client.get("/api/imported-parts/INV-SOLE", headers=headers)
+    ).status_code == 404
+
+    # AugmentedPart is archived, not deleted — direct fetch still finds it.
+    get_resp = await client.get(f"/api/augmented-parts/{ap_id}", headers=headers)
+    assert get_resp.status_code == 200
+    assert get_resp.json()["archived"] is True
+
+    # Hidden from the default list...
+    default_list = await client.get("/api/augmented-parts", headers=headers)
+    assert ap_id not in [p["id"] for p in default_list.json()]
+
+    # ...but present when archived parts are explicitly requested.
+    full_list = await client.get(
+        "/api/augmented-parts?include_archived=true", headers=headers
+    )
+    assert ap_id in [p["id"] for p in full_list.json()]
+
+
+@pytest.mark.asyncio
+async def test_patch_can_unarchive_an_augmented_part(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        ap = AugmentedPart(
+            description="Previously archived",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+            archived=True,
+        )
+        db.add(ap)
+        await db.commit()
+        ap_id = ap.id
+
+    await make_user("root2@example.com", is_superuser=True)
+    headers = {"Authorization": f"Bearer {await _token(client, 'root2@example.com')}"}
+
+    default_list = await client.get("/api/augmented-parts", headers=headers)
+    assert ap_id not in [p["id"] for p in default_list.json()]
+
+    resp = await client.patch(
+        f"/api/augmented-parts/{ap_id}", json={"archived": False}, headers=headers
+    )
+    assert resp.status_code == 200
+    assert resp.json()["archived"] is False
+
+    default_list = await client.get("/api/augmented-parts", headers=headers)
+    assert ap_id in [p["id"] for p in default_list.json()]
+
+
+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_admin_ui_hides_archived_unless_requested_and_can_toggle(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        ap = AugmentedPart(
+            description="Hidden gem",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+            archived=True,
+        )
+        db.add(ap)
+        await db.commit()
+        ap_id = ap.id
+
+    await make_user("admin@example.com", role_ids=[seeded["roles"]["admin"].id])
+    await _web_login(client, "admin@example.com")
+
+    resp = await client.get("/admin/augmented-parts")
+    assert resp.status_code == 200
+    assert "Hidden gem" not in resp.text
+
+    resp = await client.get("/admin/augmented-parts/filter?include_archived=true")
+    assert resp.status_code == 200
+    assert "Hidden gem" in resp.text
+
+    resp = await client.post(f"/admin/augmented-parts/{ap_id}/unarchive")
+    assert resp.status_code == 200
+    assert "Archived" not in resp.text
+
+    resp = await client.get("/admin/augmented-parts")
+    assert "Hidden gem" in resp.text
+
+    resp = await client.post(f"/admin/augmented-parts/{ap_id}/archive")
+    assert resp.status_code == 200
+
+    resp = await client.get("/admin/augmented-parts")
+    assert "Hidden gem" not in resp.text