Skip to content

Commit ed045cd

Browse files
jopemachineclaude
jopemachine
and
claude
committed
feat(BA-6618): app_config_fragment bulk CRUD service layer
Add bulk create / update / purge for app config fragments, stacked on the single-entity service (BA-6554). Each item reuses the allow-list write-gate; the whole batch runs in one write transaction (atomic, all-or-nothing) so a rejected gate or a missing row rolls back every item. Bulk create assigns per-config_name ranks as each row is inserted within the tx. Layers: actions (bulk_create/bulk_update/bulk_purge) + AppConfigFragmentService bulk methods + repository/db_source bulk paths (shared _*_in_tx helpers) + processors. Service layer only (no REST/GQL). Tests: service delegation (mocked) + repository real-DB bulk behavior and atomic rollback on gate rejection / missing row. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 380a8d0 commit ed045cd

10 files changed

Lines changed: 489 additions & 25 deletions

File tree

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

Lines changed: 67 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -101,8 +101,11 @@ async def _ensure_write_allowed(
101101
f"Writing app config {config_name!r} at scope {scope_type.value!r} is not allowed."
102102
)
103103

104-
@app_config_fragment_db_source_resilience.apply()
105-
async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentData:
104+
async def _create_in_tx(
105+
self, w: WriteOps, spec: AppConfigFragmentCreatorSpec
106+
) -> AppConfigFragmentData:
107+
"""Gate + create one fragment inside the caller's write transaction (rank via next-value)."""
108+
await self._ensure_write_allowed(w, spec.config_name, spec.scope_type)
106109
policy = NextValuePolicy(
107110
column=AppConfigFragmentRow.rank,
108111
scope_condition=lambda: AppConfigFragmentRow.config_name == spec.config_name,
@@ -111,10 +114,39 @@ async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentD
111114
),
112115
gap=RANK_GAP,
113116
)
117+
created = await w.create_with_next_value(policy, spec)
118+
return created.row.to_data()
119+
120+
async def _update_in_tx(
121+
self, w: WriteOps, updater: Updater[AppConfigFragmentRow]
122+
) -> AppConfigFragmentData:
123+
"""Fetch + gate + update one fragment inside the caller's write transaction."""
124+
existing = await w.query(Querier(row_class=AppConfigFragmentRow, pk_value=updater.pk_value))
125+
if existing is None:
126+
raise AppConfigFragmentNotFound(f"App config fragment {updater.pk_value} not found")
127+
await self._ensure_write_allowed(w, existing.row.config_name, existing.row.scope_type)
128+
result = await w.update(updater)
129+
if result is None:
130+
raise AppConfigFragmentNotFound(f"App config fragment {updater.pk_value} not found")
131+
return result.row.to_data()
132+
133+
async def _purge_in_tx(
134+
self, w: WriteOps, purger: Purger[AppConfigFragmentRow]
135+
) -> AppConfigFragmentData:
136+
"""Fetch + gate + purge one fragment inside the caller's write transaction."""
137+
existing = await w.query(Querier(row_class=AppConfigFragmentRow, pk_value=purger.pk_value))
138+
if existing is None:
139+
raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
140+
await self._ensure_write_allowed(w, existing.row.config_name, existing.row.scope_type)
141+
result = await w.purge(purger)
142+
if result is None:
143+
raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
144+
return result.row.to_data()
145+
146+
@app_config_fragment_db_source_resilience.apply()
147+
async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentData:
114148
async with self._ops.write_ops() as w:
115-
await self._ensure_write_allowed(w, spec.config_name, spec.scope_type)
116-
created = await w.create_with_next_value(policy, spec)
117-
return created.row.to_data()
149+
return await self._create_in_tx(w, spec)
118150

119151
@app_config_fragment_db_source_resilience.apply()
120152
async def get_by_id(self, fragment_id: AppConfigFragmentID) -> AppConfigFragmentData:
@@ -127,30 +159,40 @@ async def get_by_id(self, fragment_id: AppConfigFragmentID) -> AppConfigFragment
127159
@app_config_fragment_db_source_resilience.apply()
128160
async def update(self, updater: Updater[AppConfigFragmentRow]) -> AppConfigFragmentData:
129161
async with self._ops.write_ops() as w:
130-
existing = await w.query(
131-
Querier(row_class=AppConfigFragmentRow, pk_value=updater.pk_value)
132-
)
133-
if existing is None:
134-
raise AppConfigFragmentNotFound(f"App config fragment {updater.pk_value} not found")
135-
await self._ensure_write_allowed(w, existing.row.config_name, existing.row.scope_type)
136-
result = await w.update(updater)
137-
if result is None:
138-
raise AppConfigFragmentNotFound(f"App config fragment {updater.pk_value} not found")
139-
return result.row.to_data()
162+
return await self._update_in_tx(w, updater)
140163

141164
@app_config_fragment_db_source_resilience.apply()
142165
async def purge(self, purger: Purger[AppConfigFragmentRow]) -> AppConfigFragmentData:
143166
async with self._ops.write_ops() as w:
144-
existing = await w.query(
145-
Querier(row_class=AppConfigFragmentRow, pk_value=purger.pk_value)
146-
)
147-
if existing is None:
148-
raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
149-
await self._ensure_write_allowed(w, existing.row.config_name, existing.row.scope_type)
150-
result = await w.purge(purger)
151-
if result is None:
152-
raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
153-
return result.row.to_data()
167+
return await self._purge_in_tx(w, purger)
168+
169+
@app_config_fragment_db_source_resilience.apply()
170+
async def bulk_create(
171+
self, specs: Sequence[AppConfigFragmentCreatorSpec]
172+
) -> list[AppConfigFragmentData]:
173+
"""Create many fragments atomically — all gated and created in one transaction.
174+
175+
All-or-nothing: any rejected gate or failed insert rolls back the whole batch.
176+
Ranks are assigned per ``config_name`` as each row is inserted within the tx.
177+
"""
178+
async with self._ops.write_ops() as w:
179+
return [await self._create_in_tx(w, spec) for spec in specs]
180+
181+
@app_config_fragment_db_source_resilience.apply()
182+
async def bulk_update(
183+
self, updaters: Sequence[Updater[AppConfigFragmentRow]]
184+
) -> list[AppConfigFragmentData]:
185+
"""Update many fragments atomically — all gated and updated in one transaction."""
186+
async with self._ops.write_ops() as w:
187+
return [await self._update_in_tx(w, updater) for updater in updaters]
188+
189+
@app_config_fragment_db_source_resilience.apply()
190+
async def bulk_purge(
191+
self, purgers: Sequence[Purger[AppConfigFragmentRow]]
192+
) -> list[AppConfigFragmentData]:
193+
"""Purge many fragments atomically — all gated and purged in one transaction."""
194+
async with self._ops.write_ops() as w:
195+
return [await self._purge_in_tx(w, purger) for purger in purgers]
154196

155197
@app_config_fragment_db_source_resilience.apply()
156198
async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchResult:

‎src/ai/backend/manager/repositories/app_config_fragment/repository.py‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,3 +77,21 @@ async def scoped_search(
7777
self, querier: BatchQuerier, scopes: Sequence[SearchScope]
7878
) -> AppConfigFragmentSearchResult:
7979
return await self._db_source.scoped_search(querier, scopes)
80+
81+
@app_config_fragment_repository_resilience.apply()
82+
async def bulk_create(
83+
self, specs: Sequence[AppConfigFragmentCreatorSpec]
84+
) -> list[AppConfigFragmentData]:
85+
return await self._db_source.bulk_create(specs)
86+
87+
@app_config_fragment_repository_resilience.apply()
88+
async def bulk_update(
89+
self, updaters: Sequence[Updater[AppConfigFragmentRow]]
90+
) -> list[AppConfigFragmentData]:
91+
return await self._db_source.bulk_update(updaters)
92+
93+
@app_config_fragment_repository_resilience.apply()
94+
async def bulk_purge(
95+
self, purgers: Sequence[Purger[AppConfigFragmentRow]]
96+
) -> list[AppConfigFragmentData]:
97+
return await self._db_source.bulk_purge(purgers)

‎src/ai/backend/manager/services/app_config_fragment/actions/base.py‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
from typing import override
44

55
from ai.backend.common.data.permission.types import EntityType
6+
from ai.backend.manager.actions.action import BaseAction, BaseActionResult
67
from ai.backend.manager.actions.action.scope import BaseScopeAction, BaseScopeActionResult
78
from ai.backend.manager.actions.action.single_entity import (
89
BaseSingleEntityAction,
@@ -39,3 +40,27 @@ def field_data(self) -> FieldData | None:
3940

4041
class AppConfigFragmentSingleEntityActionResult(BaseSingleEntityActionResult):
4142
pass
43+
44+
45+
class AppConfigFragmentBulkAction(BaseAction):
46+
"""Base for bulk app config fragment mutations (bulk create / update / purge).
47+
48+
Bulk operations span many fragments (potentially across scopes), so there is no single
49+
entity id to report. Each item's write is authorized by the allow-list write-gate in the
50+
repository.
51+
"""
52+
53+
@override
54+
@classmethod
55+
def entity_type(cls) -> EntityType:
56+
return EntityType.APP_CONFIG_FRAGMENT
57+
58+
@override
59+
def entity_id(self) -> str | None:
60+
return None
61+
62+
63+
class AppConfigFragmentBulkActionResult(BaseActionResult):
64+
@override
65+
def entity_id(self) -> str | None:
66+
return None

‎src/ai/backend/manager/services/app_config_fragment/actions/bulk_create.py‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
from __future__ import annotations
2+
3+
from dataclasses import dataclass
4+
from typing import override
5+
6+
from ai.backend.manager.actions.types import ActionOperationType
7+
from ai.backend.manager.data.app_config_fragment.types import AppConfigFragmentData
8+
from ai.backend.manager.repositories.app_config_fragment.creators import (
9+
AppConfigFragmentCreatorSpec,
10+
)
11+
from ai.backend.manager.services.app_config_fragment.actions.base import (
12+
AppConfigFragmentBulkAction,
13+
AppConfigFragmentBulkActionResult,
14+
)
15+
16+
17+
@dataclass
18+
class BulkCreateAppConfigFragmentAction(AppConfigFragmentBulkAction):
19+
"""Create many fragments in a single atomic batch (all-or-nothing).
20+
21+
Each fragment is authorized by the allow-list write-gate (repository); a rejected gate or
22+
a failed insert rolls back the whole batch.
23+
"""
24+
25+
creator_specs: list[AppConfigFragmentCreatorSpec]
26+
27+
@override
28+
@classmethod
29+
def operation_type(cls) -> ActionOperationType:
30+
return ActionOperationType.CREATE
31+
32+
33+
@dataclass
34+
class BulkCreateAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult):
35+
fragments: list[AppConfigFragmentData]

‎src/ai/backend/manager/services/app_config_fragment/actions/bulk_purge.py‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
from __future__ import annotations
2+
3+
from dataclasses import dataclass
4+
from typing import override
5+
6+
from ai.backend.manager.actions.types import ActionOperationType
7+
from ai.backend.manager.data.app_config_fragment.types import AppConfigFragmentData
8+
from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
9+
from ai.backend.manager.repositories.base import Purger
10+
from ai.backend.manager.services.app_config_fragment.actions.base import (
11+
AppConfigFragmentBulkAction,
12+
AppConfigFragmentBulkActionResult,
13+
)
14+
15+
16+
@dataclass
17+
class BulkPurgeAppConfigFragmentAction(AppConfigFragmentBulkAction):
18+
"""Purge many fragments in a single atomic batch (all-or-nothing).
19+
20+
Each purge is authorized by the allow-list write-gate (repository) against the target
21+
fragment's ``(config_name, scope_type)``.
22+
"""
23+
24+
purgers: list[Purger[AppConfigFragmentRow]]
25+
26+
@override
27+
@classmethod
28+
def operation_type(cls) -> ActionOperationType:
29+
return ActionOperationType.PURGE
30+
31+
32+
@dataclass
33+
class BulkPurgeAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult):
34+
fragments: list[AppConfigFragmentData]

‎src/ai/backend/manager/services/app_config_fragment/actions/bulk_update.py‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
from __future__ import annotations
2+
3+
from dataclasses import dataclass
4+
from typing import override
5+
6+
from ai.backend.manager.actions.types import ActionOperationType
7+
from ai.backend.manager.data.app_config_fragment.types import AppConfigFragmentData
8+
from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
9+
from ai.backend.manager.repositories.base import Updater
10+
from ai.backend.manager.services.app_config_fragment.actions.base import (
11+
AppConfigFragmentBulkAction,
12+
AppConfigFragmentBulkActionResult,
13+
)
14+
15+
16+
@dataclass
17+
class BulkUpdateAppConfigFragmentAction(AppConfigFragmentBulkAction):
18+
"""Update many fragments' ``config`` in a single atomic batch (all-or-nothing).
19+
20+
Each update is authorized by the allow-list write-gate (repository) against the target
21+
fragment's ``(config_name, scope_type)``; ``rank`` is not updatable.
22+
"""
23+
24+
updaters: list[Updater[AppConfigFragmentRow]]
25+
26+
@override
27+
@classmethod
28+
def operation_type(cls) -> ActionOperationType:
29+
return ActionOperationType.UPDATE
30+
31+
32+
@dataclass
33+
class BulkUpdateAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult):
34+
fragments: list[AppConfigFragmentData]

‎src/ai/backend/manager/services/app_config_fragment/processors.py‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
from typing import override
44

55
from ai.backend.manager.actions.monitors.monitor import ActionMonitor
6+
from ai.backend.manager.actions.processor import ActionProcessor
67
from ai.backend.manager.actions.processor.bulk import BulkActionProcessor
78
from ai.backend.manager.actions.processor.scope import ScopeActionProcessor
89
from ai.backend.manager.actions.processor.single_entity import SingleEntityActionProcessor
@@ -11,6 +12,18 @@
1112
AdminSearchAppConfigFragmentAction,
1213
AdminSearchAppConfigFragmentActionResult,
1314
)
15+
from ai.backend.manager.services.app_config_fragment.actions.bulk_create import (
16+
BulkCreateAppConfigFragmentAction,
17+
BulkCreateAppConfigFragmentActionResult,
18+
)
19+
from ai.backend.manager.services.app_config_fragment.actions.bulk_purge import (
20+
BulkPurgeAppConfigFragmentAction,
21+
BulkPurgeAppConfigFragmentActionResult,
22+
)
23+
from ai.backend.manager.services.app_config_fragment.actions.bulk_update import (
24+
BulkUpdateAppConfigFragmentAction,
25+
BulkUpdateAppConfigFragmentActionResult,
26+
)
1427
from ai.backend.manager.services.app_config_fragment.actions.create import (
1528
CreateAppConfigFragmentAction,
1629
CreateAppConfigFragmentActionResult,
@@ -51,6 +64,15 @@ class AppConfigFragmentProcessors(AbstractProcessorPackage):
5164
purge: SingleEntityActionProcessor[
5265
PurgeAppConfigFragmentAction, PurgeAppConfigFragmentActionResult
5366
]
67+
bulk_create: ActionProcessor[
68+
BulkCreateAppConfigFragmentAction, BulkCreateAppConfigFragmentActionResult
69+
]
70+
bulk_update: ActionProcessor[
71+
BulkUpdateAppConfigFragmentAction, BulkUpdateAppConfigFragmentActionResult
72+
]
73+
bulk_purge: ActionProcessor[
74+
BulkPurgeAppConfigFragmentAction, BulkPurgeAppConfigFragmentActionResult
75+
]
5476

5577
def __init__(
5678
self,
@@ -63,6 +85,9 @@ def __init__(
6385
self.scoped_search = BulkActionProcessor(service.scoped_search, monitors=action_monitors)
6486
self.update = SingleEntityActionProcessor(service.update, action_monitors)
6587
self.purge = SingleEntityActionProcessor(service.purge, action_monitors)
88+
self.bulk_create = ActionProcessor(service.bulk_create, action_monitors)
89+
self.bulk_update = ActionProcessor(service.bulk_update, action_monitors)
90+
self.bulk_purge = ActionProcessor(service.bulk_purge, action_monitors)
6691

6792
@override
6893
def supported_actions(self) -> list[ActionSpec]:
@@ -73,4 +98,7 @@ def supported_actions(self) -> list[ActionSpec]:
7398
ScopedSearchAppConfigFragmentAction.spec(),
7499
UpdateAppConfigFragmentAction.spec(),
75100
PurgeAppConfigFragmentAction.spec(),
101+
BulkCreateAppConfigFragmentAction.spec(),
102+
BulkUpdateAppConfigFragmentAction.spec(),
103+
BulkPurgeAppConfigFragmentAction.spec(),
76104
]

‎src/ai/backend/manager/services/app_config_fragment/service.py‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,18 @@
77
AdminSearchAppConfigFragmentAction,
88
AdminSearchAppConfigFragmentActionResult,
99
)
10+
from ai.backend.manager.services.app_config_fragment.actions.bulk_create import (
11+
BulkCreateAppConfigFragmentAction,
12+
BulkCreateAppConfigFragmentActionResult,
13+
)
14+
from ai.backend.manager.services.app_config_fragment.actions.bulk_purge import (
15+
BulkPurgeAppConfigFragmentAction,
16+
BulkPurgeAppConfigFragmentActionResult,
17+
)
18+
from ai.backend.manager.services.app_config_fragment.actions.bulk_update import (
19+
BulkUpdateAppConfigFragmentAction,
20+
BulkUpdateAppConfigFragmentActionResult,
21+
)
1022
from ai.backend.manager.services.app_config_fragment.actions.create import (
1123
CreateAppConfigFragmentAction,
1224
CreateAppConfigFragmentActionResult,
@@ -93,3 +105,21 @@ async def purge(
93105
) -> PurgeAppConfigFragmentActionResult:
94106
data = await self._repository.purge(action.purger)
95107
return PurgeAppConfigFragmentActionResult(fragment=data)
108+
109+
async def bulk_create(
110+
self, action: BulkCreateAppConfigFragmentAction
111+
) -> BulkCreateAppConfigFragmentActionResult:
112+
fragments = await self._repository.bulk_create(action.creator_specs)
113+
return BulkCreateAppConfigFragmentActionResult(fragments=fragments)
114+
115+
async def bulk_update(
116+
self, action: BulkUpdateAppConfigFragmentAction
117+
) -> BulkUpdateAppConfigFragmentActionResult:
118+
fragments = await self._repository.bulk_update(action.updaters)
119+
return BulkUpdateAppConfigFragmentActionResult(fragments=fragments)
120+
121+
async def bulk_purge(
122+
self, action: BulkPurgeAppConfigFragmentAction
123+
) -> BulkPurgeAppConfigFragmentActionResult:
124+
fragments = await self._repository.bulk_purge(action.purgers)
125+
return BulkPurgeAppConfigFragmentActionResult(fragments=fragments)

0 commit comments

Comments
 (0)