fix(BA-5827): select only id in Fragment update/purge; expect typed FK error in tests

Mirrors the BA-5814 Policy fix (d78f510):
- `db_source.update` / `purge` now select only `AppConfigFragmentRow.id`
  when resolving the natural key, so `execute_updater`'s UPDATE ...
  RETURNING does not collide with a pre-loaded ORM instance in the
  session's identity map.
- `test_create_without_matching_policy_violates_fk` now expects
  `AppConfigFragmentPolicyMissing` — that is what the creator spec
  translates the underlying `ForeignKeyViolationError` into. Drop
  the stray `import sqlalchemy as sa` that became unused.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jopemachine

src/ai/backend/manager/repositories/app_config_fragment/db_source/db_source.py

@@ -133,18 +133,18 @@ async def update(
         Returns `None` when no row exists for the key.
         """
         async with self._db.begin_session() as db_sess:
-            row = await db_sess.scalar(
-                sa.select(AppConfigFragmentRow).where(
+            pk_value = await db_sess.scalar(
+                sa.select(AppConfigFragmentRow.id).where(
                     AppConfigFragmentRow.scope_type == key.scope_type,
                     AppConfigFragmentRow.scope_id == key.scope_id,
                     AppConfigFragmentRow.name == key.name,
                 )
             )
-            if row is None:
+            if pk_value is None:
                 return None
-            updater: Updater[AppConfigFragmentRow] = Updater(spec=spec, pk_value=row.id)
+            updater: Updater[AppConfigFragmentRow] = Updater(spec=spec, pk_value=pk_value)
             result = await execute_updater(db_sess, updater)
-            return result.row.to_data() if result is not None else row.to_data()
+            return result.row.to_data() if result is not None else None
 
     @app_config_fragment_db_source_resilience.apply()
     async def purge(self, key: AppConfigFragmentKey) -> bool:
@@ -155,17 +155,17 @@ async def purge(self, key: AppConfigFragmentKey) -> bool:
         actually removed.
         """
         async with self._db.begin_session() as db_sess:
-            row = await db_sess.scalar(
-                sa.select(AppConfigFragmentRow).where(
+            pk_value = await db_sess.scalar(
+                sa.select(AppConfigFragmentRow.id).where(
                     AppConfigFragmentRow.scope_type == key.scope_type,
                     AppConfigFragmentRow.scope_id == key.scope_id,
                     AppConfigFragmentRow.name == key.name,
                 )
             )
-            if row is None:
+            if pk_value is None:
                 return False
             purger: Purger[AppConfigFragmentRow] = Purger(
-                row_class=AppConfigFragmentRow, pk_value=row.id
+                row_class=AppConfigFragmentRow, pk_value=pk_value
             )
             result = await execute_purger(db_sess, purger)
             return result is not None

tests/unit/manager/repositories/app_config_fragment/test_app_config_fragment.py

@@ -5,12 +5,12 @@
 from collections.abc import AsyncGenerator
 
 import pytest
-import sqlalchemy as sa
 
 from ai.backend.manager.data.app_config_fragment.types import (
     AppConfigFragmentKey,
     AppConfigScopeType,
 )
+from ai.backend.manager.errors.app_config import AppConfigFragmentPolicyMissing
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.app_config_policy.row import AppConfigPolicyRow
 from ai.backend.manager.models.utils import ExtendedAsyncSAEngine
@@ -221,8 +221,10 @@ async def test_create_without_matching_policy_violates_fk(
         admin_repository: AppConfigFragmentAdminRepository,
     ) -> None:
         # Required-policy invariant (BEP-1052 §1) — DB-level FK is the
-        # backstop when the service layer's check is bypassed.
-        with pytest.raises(sa.exc.IntegrityError):
+        # backstop when the service layer's check is bypassed. The
+        # creator spec translates the `ForeignKeyViolationError` into
+        # the typed `AppConfigFragmentPolicyMissing` domain error.
+        with pytest.raises(AppConfigFragmentPolicyMissing):
             await admin_repository.create(
                 key=AppConfigFragmentKey(
                     scope_type=AppConfigScopeType.DOMAIN,