Adding ability to update / patch by sku, adding uniqueness constraint violation checks

alembic/versions/1c2d3e4f5a6b_augmented_parts_sku_unique.py

@@ -0,0 +1,26 @@
+"""augmented_parts_sku_unique
+
+Revision ID: 1c2d3e4f5a6b
+Revises: 0fec09fb6ae6
+Create Date: 2026-09-21 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "1c2d3e4f5a6b"
+down_revision: str | None = "0fec09fb6ae6"
+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.create_index("ix_augmented_parts_sku", ["sku"], unique=True)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("augmented_parts", schema=None) as batch_op:
+        batch_op.drop_index("ix_augmented_parts_sku")

app/api/attachment_point_names.py

@@ -33,6 +33,10 @@ async def create_attachment_point_name(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("attachment_point_names:write")),
 ):
+    if await apn_crud.get_by_name(db, payload.name):
+        raise HTTPException(
+            status_code=409, detail="Attachment point name already exists"
+        )
     apn = await apn_crud.create(db, name=payload.name)
     await audit_crud.log(
         db,
@@ -68,6 +72,12 @@ async def update_attachment_point_name(
     if apn is None:
         raise HTTPException(status_code=404, detail="Attachment point name not found")
     changes = payload.model_dump(exclude_unset=True)
+    if "name" in changes:
+        existing = await apn_crud.get_by_name(db, changes["name"])
+        if existing is not None and existing.id != apn.id:
+            raise HTTPException(
+                status_code=409, detail="Attachment point name already exists"
+            )
     apn = await apn_crud.update(db, apn, changes)
     await audit_crud.log(
         db,

app/api/augmented_parts.py

@@ -20,6 +20,15 @@ from app.services import augmented_part_sync
 router = APIRouter()
 
 
+async def _get_augmented_part_by_identifier(db: AsyncSession, identifier: str):
+    """Resolve an AugmentedPart by numeric id, falling back to sku."""
+    if identifier.isdigit():
+        ap = await augmented_part_crud.get(db, int(identifier))
+        if ap is not None:
+            return ap
+    return await augmented_part_crud.get_by_sku(db, identifier)
+
+
 @router.get("", response_model=list[AugmentedPartRead])
 async def list_augmented_parts(
     skip: int = 0,
@@ -39,6 +48,8 @@ async def create_augmented_part(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("augmented_parts:write")),
 ):
+    if payload.sku and await augmented_part_crud.get_by_sku(db, payload.sku):
+        raise HTTPException(status_code=409, detail="SKU already in use")
     ap = await augmented_part_crud.create(
         db,
         sku=payload.sku,
@@ -94,17 +105,21 @@ async def get_augmented_part(
     return ap
 
 
-@router.patch("/{augmented_part_id}", response_model=AugmentedPartRead)
+@router.patch("/{identifier}", response_model=AugmentedPartRead)
 async def update_augmented_part(
-    augmented_part_id: int,
+    identifier: str,
     payload: AugmentedPartUpdate,
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("augmented_parts:write")),
 ):
-    ap = await augmented_part_crud.get(db, augmented_part_id)
+    ap = await _get_augmented_part_by_identifier(db, identifier)
     if ap is None:
         raise HTTPException(status_code=404, detail="Augmented part not found")
     changes = payload.model_dump(exclude_unset=True)
+    if changes.get("sku"):
+        existing = await augmented_part_crud.get_by_sku(db, changes["sku"])
+        if existing is not None and existing.id != ap.id:
+            raise HTTPException(status_code=409, detail="SKU already in use")
     ap = await augmented_part_crud.update(db, ap, changes)
     await audit_crud.log(
         db,
@@ -117,15 +132,16 @@ async def update_augmented_part(
     return ap
 
 
-@router.delete("/{augmented_part_id}", status_code=status.HTTP_204_NO_CONTENT)
+@router.delete("/{identifier}", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_augmented_part(
-    augmented_part_id: int,
+    identifier: str,
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("augmented_parts:delete")),
 ):
-    ap = await augmented_part_crud.get(db, augmented_part_id)
+    ap = await _get_augmented_part_by_identifier(db, identifier)
     if ap is None:
         raise HTTPException(status_code=404, detail="Augmented part not found")
+    resolved_id = ap.id
     label = ap.sku or ap.description
     await augmented_part_crud.delete(db, ap)
     await audit_crud.log(
@@ -133,10 +149,8 @@ async def delete_augmented_part(
         actor=actor,
         action="delete",
         resource_type="AugmentedPart",
-        resource_id=augmented_part_id,
+        resource_id=resolved_id,
-        description=audit_crud.describe_delete(
+        description=audit_crud.describe_delete("AugmentedPart", resolved_id, label),
-            "AugmentedPart", augmented_part_id, label
-        ),
     )
 
 

app/api/part_colours.py

@@ -27,6 +27,8 @@ async def create_part_colour(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("part_colours:write")),
 ):
+    if await part_colour_crud.get_by_name(db, payload.name):
+        raise HTTPException(status_code=409, detail="Part colour already exists")
     pc = await part_colour_crud.create(db, name=payload.name)
     await audit_crud.log(
         db,
@@ -62,6 +64,10 @@ async def update_part_colour(
     if pc is None:
         raise HTTPException(status_code=404, detail="Part colour not found")
     changes = payload.model_dump(exclude_unset=True)
+    if "name" in changes:
+        existing = await part_colour_crud.get_by_name(db, changes["name"])
+        if existing is not None and existing.id != pc.id:
+            raise HTTPException(status_code=409, detail="Part colour already exists")
     pc = await part_colour_crud.update(db, pc, changes)
     await audit_crud.log(
         db,

app/api/part_types.py

@@ -27,6 +27,10 @@ async def create_part_type(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("part_types:write")),
 ):
+    if await part_type_crud.get_by_name(db, payload.name):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Part type already exists"
+        )
     pt = await part_type_crud.create(db, name=payload.name)
     await audit_crud.log(
         db,
@@ -66,6 +70,12 @@ async def update_part_type(
             status_code=status.HTTP_404_NOT_FOUND, detail="Part type not found"
         )
     changes = payload.model_dump(exclude_unset=True)
+    if "name" in changes:
+        existing = await part_type_crud.get_by_name(db, changes["name"])
+        if existing is not None and existing.id != pt.id:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail="Part type already exists"
+            )
     pt = await part_type_crud.update(db, pt, changes)
     await audit_crud.log(
         db,

app/api/roles.py

@@ -81,6 +81,12 @@ async def update_role(
             status_code=status.HTTP_404_NOT_FOUND, detail="Role not found"
         )
     changes = payload.model_dump(exclude_unset=True)
+    if "name" in changes:
+        existing = await role_crud.get_by_name(db, changes["name"])
+        if existing is not None and existing.id != role.id:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail="Role already exists"
+            )
     role = await role_crud.update(db, role, changes)
     await audit_crud.log(
         db,

app/api/static_connections.py

@@ -33,6 +33,8 @@ async def create_static_connection(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("static_connections:write")),
 ):
+    if await static_connection_crud.get_by_od_mm(db, payload.od_mm):
+        raise HTTPException(status_code=409, detail="Static connection already exists")
     sc = await static_connection_crud.create(
         db, od_mm=payload.od_mm, od_in=payload.od_in
     )
@@ -72,6 +74,12 @@ async def update_static_connection(
     if sc is None:
         raise HTTPException(status_code=404, detail="Static connection not found")
     changes = payload.model_dump(exclude_unset=True)
+    if "od_mm" in changes:
+        existing = await static_connection_crud.get_by_od_mm(db, changes["od_mm"])
+        if existing is not None and existing.id != sc.id:
+            raise HTTPException(
+                status_code=409, detail="Static connection already exists"
+            )
     sc = await static_connection_crud.update(db, sc, changes)
     await audit_crud.log(
         db,

app/api/sub_part_types.py

@@ -31,6 +31,10 @@ async def create_sub_part_type(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(require_permission("sub_part_types:write")),
 ):
+    if await sub_part_type_crud.get_by_name_and_part_type(
+        db, payload.name, payload.part_type_id
+    ):
+        raise HTTPException(status_code=409, detail="Sub part type already exists")
     spt = await sub_part_type_crud.create(
         db, name=payload.name, part_type_id=payload.part_type_id
     )
@@ -72,6 +76,14 @@ async def update_sub_part_type(
             status_code=status.HTTP_404_NOT_FOUND, detail="Sub part type not found"
         )
     changes = payload.model_dump(exclude_unset=True)
+    if "name" in changes or "part_type_id" in changes:
+        name = changes.get("name", spt.name)
+        part_type_id = changes.get("part_type_id", spt.part_type_id)
+        existing = await sub_part_type_crud.get_by_name_and_part_type(
+            db, name, part_type_id
+        )
+        if existing is not None and existing.id != spt.id:
+            raise HTTPException(status_code=409, detail="Sub part type already exists")
     spt = await sub_part_type_crud.update(db, spt, changes)
     await audit_crud.log(
         db,

app/api/users.py

@@ -82,6 +82,12 @@ async def update_user(
             status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
         )
     changes = payload.model_dump(exclude_unset=True)
+    if "email" in changes:
+        existing = await user_crud.get_by_email(db, changes["email"])
+        if existing is not None and existing.id != target.id:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail="Email already registered"
+            )
     target = await user_crud.update(db, target, changes)
     await audit_crud.log(
         db,

app/models/augmented_part.py

@@ -17,7 +17,9 @@ class AugmentedPart(Base):
     __tablename__ = "augmented_parts"
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    sku: Mapped[str | None] = mapped_column(String, nullable=True, default=None)
+    sku: Mapped[str | None] = mapped_column(
+        String, nullable=True, unique=True, default=None
+    )
     description: Mapped[str] = mapped_column(String, nullable=False)
     colour_id: Mapped[int] = mapped_column(
         ForeignKey("part_colours.id", ondelete="RESTRICT"), nullable=False

app/web/admin.py

@@ -1047,6 +1047,9 @@ async def update_part_type(
     pt = await part_type_crud.get(db, part_type_id)
     if pt is None:
         raise HTTPException(status_code=404, detail="Part type not found")
+    existing = await part_type_crud.get_by_name(db, name)
+    if existing is not None and existing.id != pt.id:
+        raise HTTPException(status_code=409, detail="Part type already exists")
     changes = {"name": name}
     await part_type_crud.update(db, pt, changes)
     await audit_crud.log(
@@ -1174,6 +1177,11 @@ async def update_sub_part_type(
     spt = await sub_part_type_crud.get(db, sub_part_type_id)
     if spt is None:
         raise HTTPException(status_code=404, detail="Sub part type not found")
+    existing = await sub_part_type_crud.get_by_name_and_part_type(
+        db, name, part_type_id
+    )
+    if existing is not None and existing.id != spt.id:
+        raise HTTPException(status_code=409, detail="Sub part type already exists")
     changes = {"name": name, "part_type_id": part_type_id}
     await sub_part_type_crud.update(db, spt, changes)
     await audit_crud.log(
@@ -1724,9 +1732,12 @@ async def create_augmented_part(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(web_require("augmented_parts:write")),
 ):
+    clean_sku = sku.strip() or None
+    if clean_sku and await augmented_part_crud.get_by_sku(db, clean_sku):
+        raise HTTPException(status_code=409, detail="SKU already in use")
     created = await augmented_part_crud.create(
         db,
-        sku=sku.strip() or None,
+        sku=clean_sku,
         description=description,
         colour_id=colour_id,
         manufacturer_id=manufacturer_id,
@@ -2180,8 +2191,13 @@ async def update_augmented_part(
     ap = await augmented_part_crud.get(db, augmented_part_id)
     if ap is None:
         raise HTTPException(status_code=404, detail="Augmented part not found")
+    clean_sku = sku.strip() or None
+    if clean_sku:
+        existing = await augmented_part_crud.get_by_sku(db, clean_sku)
+        if existing is not None and existing.id != ap.id:
+            raise HTTPException(status_code=409, detail="SKU already in use")
     changes = {
-        "sku": sku.strip() or None,
+        "sku": clean_sku,
         "description": description,
         "colour_id": colour_id,
         "manufacturer_id": manufacturer_id,
@@ -3065,6 +3081,9 @@ async def update_part_colour(
     pc = await part_colour_crud.get(db, part_colour_id)
     if pc is None:
         raise HTTPException(status_code=404, detail="Part colour not found")
+    existing = await part_colour_crud.get_by_name(db, name)
+    if existing is not None and existing.id != pc.id:
+        raise HTTPException(status_code=409, detail="Part colour already exists")
     changes = {"name": name}
     await part_colour_crud.update(db, pc, changes)
     await audit_crud.log(
@@ -3184,6 +3203,11 @@ async def update_attachment_point_name(
     apn = await apn_crud.get(db, apn_id)
     if apn is None:
         raise HTTPException(status_code=404, detail="Attachment point name not found")
+    existing = await apn_crud.get_by_name(db, name)
+    if existing is not None and existing.id != apn.id:
+        raise HTTPException(
+            status_code=409, detail="Attachment point name already exists"
+        )
     changes = {"name": name}
     await apn_crud.update(db, apn, changes)
     await audit_crud.log(
@@ -3521,6 +3545,9 @@ async def update_static_connection(
     sc = await static_connection_crud.get(db, static_connection_id)
     if sc is None:
         raise HTTPException(status_code=404, detail="Static connection not found")
+    existing = await static_connection_crud.get_by_od_mm(db, od_mm)
+    if existing is not None and existing.id != sc.id:
+        raise HTTPException(status_code=409, detail="Static connection already exists")
     changes = {"od_mm": od_mm, "od_in": od_in}
     await static_connection_crud.update(db, sc, changes)
     await audit_crud.log(

tests/test_augmented_part_archiving.py

@@ -210,6 +210,86 @@ async def test_patch_can_unarchive_an_augmented_part(
     assert ap_id in [p["id"] for p in default_list.json()]
 
 
+@pytest.mark.asyncio
+async def test_patch_and_delete_accept_sku_as_identifier(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        ap = AugmentedPart(
+            description="Sku addressed part",
+            sku="SKU-123",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+        )
+        db.add(ap)
+        await db.commit()
+        ap_id = ap.id
+
+    await make_user("root3@example.com", is_superuser=True)
+    headers = {"Authorization": f"Bearer {await _token(client, 'root3@example.com')}"}
+
+    resp = await client.patch(
+        "/api/augmented-parts/SKU-123",
+        json={"description": "Renamed via sku"},
+        headers=headers,
+    )
+    assert resp.status_code == 200
+    assert resp.json()["id"] == ap_id
+    assert resp.json()["description"] == "Renamed via sku"
+
+    resp = await client.delete("/api/augmented-parts/SKU-123", headers=headers)
+    assert resp.status_code == 204
+
+    assert (
+        await client.get(f"/api/augmented-parts/{ap_id}", headers=headers)
+    ).status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_delete_by_id_still_works(client, seeded, make_user, db_sessionmaker):
+    async with db_sessionmaker() as db:
+        manufacturer, colour, part_type = await _seed_reference_data(db)
+        ap = AugmentedPart(
+            description="Delete me by id",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+        )
+        db.add(ap)
+        await db.commit()
+        ap_id = ap.id
+
+    await make_user("root4@example.com", is_superuser=True)
+    headers = {"Authorization": f"Bearer {await _token(client, 'root4@example.com')}"}
+
+    resp = await client.delete(f"/api/augmented-parts/{ap_id}", headers=headers)
+    assert resp.status_code == 204
+
+    assert (
+        await client.get(f"/api/augmented-parts/{ap_id}", headers=headers)
+    ).status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_update_and_delete_by_unknown_sku_404s(
+    client, seeded, make_user, db_sessionmaker
+):
+    await make_user("root5@example.com", is_superuser=True)
+    headers = {"Authorization": f"Bearer {await _token(client, 'root5@example.com')}"}
+
+    resp = await client.patch(
+        "/api/augmented-parts/NO-SUCH-SKU",
+        json={"description": "won't happen"},
+        headers=headers,
+    )
+    assert resp.status_code == 404
+
+    resp = await client.delete("/api/augmented-parts/NO-SUCH-SKU", headers=headers)
+    assert resp.status_code == 404
+
+
 async def _web_login(client, email, password="secret123"):
     resp = await client.post("/login", data={"email": email, "password": password})
     assert resp.status_code == 303

tests/test_duplicate_constraint_handling.py

@@ -0,0 +1,321 @@
+"""Duplicate-value pre-checks (409) for every unique-constrained field, across
+both the JSON API (app/api/*.py) and the HTMX admin UI (app/web/admin.py)."""
+
+import pytest
+
+
+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"]
+
+
+async def _superuser_headers(client, make_user, email):
+    await make_user(email, is_superuser=True)
+    return {"Authorization": f"Bearer {await _token(client, email)}"}
+
+
+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_part_type_duplicate_name_409_on_create_and_update(
+    client, seeded, make_user
+):
+    headers = await _superuser_headers(client, make_user, "pt-api@example.com")
+
+    resp = await client.post("/api/part-types", json={"name": "Scope"}, headers=headers)
+    assert resp.status_code == 201
+    other = (
+        await client.post("/api/part-types", json={"name": "Cover"}, headers=headers)
+    ).json()
+
+    resp = await client.post("/api/part-types", json={"name": "Scope"}, headers=headers)
+    assert resp.status_code == 409
+
+    resp = await client.patch(
+        f"/api/part-types/{other['id']}", json={"name": "Scope"}, headers=headers
+    )
+    assert resp.status_code == 409
+
+    # Renaming to the resource's own current name is not a false-positive collision.
+    resp = await client.patch(
+        f"/api/part-types/{other['id']}", json={"name": "Cover"}, headers=headers
+    )
+    assert resp.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_part_colour_duplicate_name_409_on_create_and_update(
+    client, seeded, make_user
+):
+    headers = await _superuser_headers(client, make_user, "pc-api@example.com")
+
+    assert (
+        await client.post("/api/part-colours", json={"name": "Black"}, headers=headers)
+    ).status_code == 201
+    other = (
+        await client.post("/api/part-colours", json={"name": "Tan"}, headers=headers)
+    ).json()
+
+    resp = await client.post(
+        "/api/part-colours", json={"name": "Black"}, headers=headers
+    )
+    assert resp.status_code == 409
+
+    resp = await client.patch(
+        f"/api/part-colours/{other['id']}", json={"name": "Black"}, headers=headers
+    )
+    assert resp.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_attachment_point_name_duplicate_409_on_create_and_update(
+    client, seeded, make_user
+):
+    headers = await _superuser_headers(client, make_user, "apn-api@example.com")
+
+    assert (
+        await client.post(
+            "/api/attachment-point-names", json={"name": "Ocular"}, headers=headers
+        )
+    ).status_code == 201
+    other = (
+        await client.post(
+            "/api/attachment-point-names", json={"name": "Objective"}, headers=headers
+        )
+    ).json()
+
+    resp = await client.post(
+        "/api/attachment-point-names", json={"name": "Ocular"}, headers=headers
+    )
+    assert resp.status_code == 409
+
+    resp = await client.patch(
+        f"/api/attachment-point-names/{other['id']}",
+        json={"name": "Ocular"},
+        headers=headers,
+    )
+    assert resp.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_static_connection_duplicate_od_mm_409_on_create_and_update(
+    client, seeded, make_user
+):
+    headers = await _superuser_headers(client, make_user, "sc-api@example.com")
+
+    assert (
+        await client.post(
+            "/api/static-connections",
+            json={"od_mm": 40.0, "od_in": 1.575},
+            headers=headers,
+        )
+    ).status_code == 201
+    other = (
+        await client.post(
+            "/api/static-connections",
+            json={"od_mm": 56.0, "od_in": 2.205},
+            headers=headers,
+        )
+    ).json()
+
+    resp = await client.post(
+        "/api/static-connections", json={"od_mm": 40.0, "od_in": 1.6}, headers=headers
+    )
+    assert resp.status_code == 409
+
+    resp = await client.patch(
+        f"/api/static-connections/{other['id']}",
+        json={"od_mm": 40.0},
+        headers=headers,
+    )
+    assert resp.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_sub_part_type_duplicate_composite_409_on_create_and_update(
+    client, seeded, make_user
+):
+    headers = await _superuser_headers(client, make_user, "spt-api@example.com")
+
+    part_type = (
+        await client.post("/api/part-types", json={"name": "Scope"}, headers=headers)
+    ).json()
+
+    assert (
+        await client.post(
+            "/api/sub-part-types",
+            json={"name": "FFP", "part_type_id": part_type["id"]},
+            headers=headers,
+        )
+    ).status_code == 201
+    other = (
+        await client.post(
+            "/api/sub-part-types",
+            json={"name": "SFP", "part_type_id": part_type["id"]},
+            headers=headers,
+        )
+    ).json()
+
+    resp = await client.post(
+        "/api/sub-part-types",
+        json={"name": "FFP", "part_type_id": part_type["id"]},
+        headers=headers,
+    )
+    assert resp.status_code == 409
+
+    resp = await client.patch(
+        f"/api/sub-part-types/{other['id']}", json={"name": "FFP"}, headers=headers
+    )
+    assert resp.status_code == 409
+
+    # Same name, but a *different* part_type_id, is not a collision.
+    other_part_type = (
+        await client.post("/api/part-types", json={"name": "Cover"}, headers=headers)
+    ).json()
+    resp = await client.post(
+        "/api/sub-part-types",
+        json={"name": "FFP", "part_type_id": other_part_type["id"]},
+        headers=headers,
+    )
+    assert resp.status_code == 201
+
+
+@pytest.mark.asyncio
+async def test_role_duplicate_name_409_on_create_and_update(client, seeded, make_user):
+    headers = await _superuser_headers(client, make_user, "role-api@example.com")
+
+    assert (
+        await client.post("/api/roles", json={"name": "editors"}, headers=headers)
+    ).status_code == 201
+    other = (
+        await client.post("/api/roles", json={"name": "viewers"}, headers=headers)
+    ).json()
+
+    resp = await client.post("/api/roles", json={"name": "editors"}, headers=headers)
+    assert resp.status_code == 409
+
+    resp = await client.patch(
+        f"/api/roles/{other['id']}", json={"name": "editors"}, headers=headers
+    )
+    assert resp.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_user_duplicate_email_409_on_create_and_update(client, seeded, make_user):
+    headers = await _superuser_headers(client, make_user, "user-api@example.com")
+
+    assert (
+        await client.post(
+            "/api/users",
+            json={"email": "dup@example.com", "password": "secret123"},
+            headers=headers,
+        )
+    ).status_code == 201
+    other = (
+        await client.post(
+            "/api/users",
+            json={"email": "other@example.com", "password": "secret123"},
+            headers=headers,
+        )
+    ).json()
+
+    resp = await client.post(
+        "/api/users",
+        json={"email": "dup@example.com", "password": "secret123"},
+        headers=headers,
+    )
+    assert resp.status_code == 409
+
+    resp = await client.patch(
+        f"/api/users/{other['id']}", json={"email": "dup@example.com"}, headers=headers
+    )
+    assert resp.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_admin_ui_update_part_type_duplicate_name_409(
+    client, seeded, make_user, db_sessionmaker
+):
+    from app.crud import part_type as part_type_crud
+
+    async with db_sessionmaker() as db:
+        await part_type_crud.create(db, name="Scope")
+        cover = await part_type_crud.create(db, name="Cover")
+        cover_id = cover.id
+
+    await make_user("pt-web@example.com", role_ids=[seeded["roles"]["admin"].id])
+    await _web_login(client, "pt-web@example.com")
+
+    resp = await client.post(f"/admin/part-types/{cover_id}", data={"name": "Scope"})
+    assert resp.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_admin_ui_augmented_part_duplicate_sku_409_on_create_and_update(
+    client, seeded, make_user, db_sessionmaker
+):
+    from app.models.augmented_part import AugmentedPart
+    from app.models.manufacturer import Manufacturer
+    from app.models.part_colour import PartColour
+    from app.models.part_type import PartType
+
+    async with db_sessionmaker() as 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()
+        existing = AugmentedPart(
+            description="Taken sku part",
+            sku="DUP-1",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+        )
+        other = AugmentedPart(
+            description="Other part",
+            colour_id=colour.id,
+            manufacturer_id=manufacturer.id,
+            part_type_id=part_type.id,
+        )
+        db.add_all([existing, other])
+        await db.commit()
+        manufacturer_id, colour_id, part_type_id = (
+            manufacturer.id,
+            colour.id,
+            part_type.id,
+        )
+        other_id = other.id
+
+    await make_user("ap-web@example.com", role_ids=[seeded["roles"]["admin"].id])
+    await _web_login(client, "ap-web@example.com")
+
+    resp = await client.post(
+        "/admin/augmented-parts",
+        data={
+            "sku": "DUP-1",
+            "description": "New part",
+            "colour_id": colour_id,
+            "manufacturer_id": manufacturer_id,
+            "part_type_id": part_type_id,
+        },
+    )
+    assert resp.status_code == 409
+
+    resp = await client.post(
+        f"/admin/augmented-parts/{other_id}",
+        data={
+            "sku": "DUP-1",
+            "description": "Other part",
+            "colour_id": colour_id,
+            "manufacturer_id": manufacturer_id,
+            "part_type_id": part_type_id,
+        },
+    )
+    assert resp.status_code == 409