Skip to content

Commit ab3e072

Browse files
jopemachineclaude
andcommitted
feat(BA-6626): app_config_fragment bulk repository layer
Build the app_config_fragment bulk repository on the conditional-bulk primitives (#12429): - bulk_create / bulk_update / bulk_purge in db_source + repository, returning AppConfigFragmentBulkWriteResult (succeeded + failed[index, message]) — partial success via the WriteOps.bulk_*_partial primitives. - bulk_create takes plain Creators; the FK to the allow-list is the gate, so an item with no allow-list row fails per-item as AppConfigFragmentWriteNotAllowed (via the spec's integrity checks) while the rest are created. - Repository unit tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent e74d747 commit ab3e072

5 files changed

Lines changed: 291 additions & 1 deletion

File tree

changes/12426.feature.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Add `app_config_fragment` bulk repository operations (`bulk_create`/`bulk_update`/`bulk_purge` with per-item partial success): bulk create relies on the allow-list FK so an item with no allow-list row is rejected per item, bulk update/purge report missing targets per item, and the `AppConfigFragmentBulkWriteResult` data type carries the partial result

src/ai/backend/manager/data/app_config_fragment/types.py

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,3 +29,23 @@ class AppConfigFragmentSearchResult:
2929
total_count: int
3030
has_next_page: bool
3131
has_previous_page: bool
32+
33+
34+
@dataclass(frozen=True)
35+
class AppConfigFragmentBulkItemError:
36+
"""One failed item of a partial bulk mutation: its batch position and a reason."""
37+
38+
index: int
39+
message: str
40+
41+
42+
@dataclass(frozen=True)
43+
class AppConfigFragmentBulkWriteResult:
44+
"""Partial-success result of a bulk mutation.
45+
46+
``succeeded`` are the fragments that were created/updated/purged; ``failed`` are the items
47+
whose gate was rejected or whose write failed, each with its batch ``index`` and a reason.
48+
"""
49+
50+
succeeded: list[AppConfigFragmentData]
51+
failed: list[AppConfigFragmentBulkItemError]

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

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@
1313
from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
1414
from ai.backend.common.resilience.resilience import Resilience
1515
from ai.backend.manager.data.app_config_fragment.types import (
16+
AppConfigFragmentBulkItemError,
17+
AppConfigFragmentBulkWriteResult,
1618
AppConfigFragmentData,
1719
AppConfigFragmentSearchResult,
1820
)
@@ -23,6 +25,7 @@
2325
from ai.backend.manager.models.scopes import SearchScope
2426
from ai.backend.manager.repositories.base import (
2527
BatchQuerier,
28+
BulkCreator,
2629
Creator,
2730
Purger,
2831
Querier,
@@ -97,6 +100,70 @@ async def purge(self, purger: Purger[AppConfigFragmentRow]) -> AppConfigFragment
97100
raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
98101
return result.row.to_data()
99102

103+
@app_config_fragment_db_source_resilience.apply()
104+
async def bulk_create(
105+
self,
106+
bulk_creator: BulkCreator[AppConfigFragmentRow],
107+
) -> AppConfigFragmentBulkWriteResult:
108+
"""Create many fragments with per-item partial success."""
109+
async with self._ops.write_ops() as w:
110+
result = await w.bulk_create_partial(bulk_creator)
111+
return AppConfigFragmentBulkWriteResult(
112+
succeeded=[row.to_data() for row in result.successes],
113+
failed=[
114+
AppConfigFragmentBulkItemError(index=error.index, message=str(error.exception))
115+
for error in result.errors
116+
],
117+
)
118+
119+
@app_config_fragment_db_source_resilience.apply()
120+
async def bulk_update(
121+
self,
122+
updaters: Sequence[Updater[AppConfigFragmentRow]],
123+
) -> AppConfigFragmentBulkWriteResult:
124+
"""Update many fragments with per-item partial success."""
125+
async with self._ops.write_ops() as w:
126+
result = await w.bulk_update_partial(updaters)
127+
succeeded = [row.to_data() for row in result.successes]
128+
succeeded_ids = {data.id for data in succeeded}
129+
errors_by_index = {e.index: str(e.exception) for e in result.errors}
130+
# A missing PK is skipped by the partial op (no row, no error); report as not-found.
131+
failed = [
132+
AppConfigFragmentBulkItemError(
133+
index=index,
134+
message=errors_by_index.get(
135+
index, f"App config fragment {updater.pk_value} not found"
136+
),
137+
)
138+
for index, updater in enumerate(updaters)
139+
if index in errors_by_index or updater.pk_value not in succeeded_ids
140+
]
141+
return AppConfigFragmentBulkWriteResult(succeeded=succeeded, failed=failed)
142+
143+
@app_config_fragment_db_source_resilience.apply()
144+
async def bulk_purge(
145+
self,
146+
purgers: Sequence[Purger[AppConfigFragmentRow]],
147+
) -> AppConfigFragmentBulkWriteResult:
148+
"""Purge many fragments with per-item partial success."""
149+
async with self._ops.write_ops() as w:
150+
result = await w.bulk_purge_partial(list(purgers))
151+
succeeded = [row.to_data() for row in result.successes]
152+
succeeded_ids = {data.id for data in succeeded}
153+
errors_by_index = {e.index: str(e.exception) for e in result.errors}
154+
# A missing PK is skipped by the partial op (no row, no error); report as not-found.
155+
failed = [
156+
AppConfigFragmentBulkItemError(
157+
index=index,
158+
message=errors_by_index.get(
159+
index, f"App config fragment {purger.pk_value} not found"
160+
),
161+
)
162+
for index, purger in enumerate(purgers)
163+
if index in errors_by_index or purger.pk_value not in succeeded_ids
164+
]
165+
return AppConfigFragmentBulkWriteResult(succeeded=succeeded, failed=failed)
166+
100167
@app_config_fragment_db_source_resilience.apply()
101168
async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchResult:
102169
"""Superadmin/internal path: query across all fragments with no scope filter."""

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

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
1010
from ai.backend.common.resilience.resilience import Resilience
1111
from ai.backend.manager.data.app_config_fragment.types import (
12+
AppConfigFragmentBulkWriteResult,
1213
AppConfigFragmentData,
1314
AppConfigFragmentSearchResult,
1415
)
@@ -17,7 +18,13 @@
1718
from ai.backend.manager.repositories.app_config_fragment.db_source import (
1819
AppConfigFragmentDBSource,
1920
)
20-
from ai.backend.manager.repositories.base import BatchQuerier, Creator, Purger, Updater
21+
from ai.backend.manager.repositories.base import (
22+
BatchQuerier,
23+
BulkCreator,
24+
Creator,
25+
Purger,
26+
Updater,
27+
)
2128
from ai.backend.manager.repositories.ops import DBOpsProvider
2229

2330
__all__ = ("AppConfigFragmentRepository",)
@@ -75,3 +82,24 @@ async def scoped_search(
7582
self, querier: BatchQuerier, scopes: Sequence[SearchScope]
7683
) -> AppConfigFragmentSearchResult:
7784
return await self._db_source.scoped_search(querier, scopes)
85+
86+
@app_config_fragment_repository_resilience.apply()
87+
async def bulk_create(
88+
self,
89+
bulk_creator: BulkCreator[AppConfigFragmentRow],
90+
) -> AppConfigFragmentBulkWriteResult:
91+
return await self._db_source.bulk_create(bulk_creator)
92+
93+
@app_config_fragment_repository_resilience.apply()
94+
async def bulk_update(
95+
self,
96+
updaters: Sequence[Updater[AppConfigFragmentRow]],
97+
) -> AppConfigFragmentBulkWriteResult:
98+
return await self._db_source.bulk_update(updaters)
99+
100+
@app_config_fragment_repository_resilience.apply()
101+
async def bulk_purge(
102+
self,
103+
purgers: Sequence[Purger[AppConfigFragmentRow]],
104+
) -> AppConfigFragmentBulkWriteResult:
105+
return await self._db_source.bulk_purge(purgers)

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

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
)
4040
from ai.backend.manager.repositories.base import (
4141
BatchQuerier,
42+
BulkCreator,
4243
Creator,
4344
OffsetPagination,
4445
Purger,
@@ -397,3 +398,176 @@ async def test_scoped_search_unknown_scope_returns_empty(
397398
[DomainAppConfigFragmentSearchScope(domain_id=DomainID(uuid.uuid4()))],
398399
)
399400
assert result.items == []
401+
402+
403+
@pytest.fixture
404+
async def menu_defined(database: ExtendedAsyncSAEngine, theme_registered: None) -> None:
405+
"""``theme`` is allow-listed at every scope (theme_registered); ``menu`` is defined, NOT allow-listed."""
406+
async with database.begin_session() as db_sess:
407+
db_sess.add(AppConfigDefinitionRow(config_name="menu"))
408+
await db_sess.flush()
409+
410+
411+
@pytest.fixture
412+
async def two_fragments(database: ExtendedAsyncSAEngine) -> list[AppConfigFragmentData]:
413+
"""A domain-scoped and a user-scoped ``theme`` fragment, both allow-listed."""
414+
async with database.begin_session() as db_sess:
415+
db_sess.add(AppConfigDefinitionRow(config_name="theme"))
416+
await db_sess.flush()
417+
db_sess.add_all([
418+
_allow_list_row("theme", AppConfigScopeType.DOMAIN),
419+
_allow_list_row("theme", AppConfigScopeType.USER),
420+
])
421+
await db_sess.flush()
422+
rows = [
423+
AppConfigFragmentRow(
424+
config_name="theme",
425+
scope_type=AppConfigScopeType.DOMAIN,
426+
scope_id=_DOMAIN_ID,
427+
config={"a": 1},
428+
),
429+
AppConfigFragmentRow(
430+
config_name="theme",
431+
scope_type=AppConfigScopeType.USER,
432+
scope_id=_USER_ID,
433+
config={"b": 2},
434+
),
435+
]
436+
db_sess.add_all(rows)
437+
await db_sess.flush()
438+
return [row.to_data() for row in rows]
439+
440+
441+
class TestBulkCreate:
442+
async def test_all_created(
443+
self, repository: AppConfigFragmentRepository, theme_registered: None
444+
) -> None:
445+
result = await repository.bulk_create(
446+
BulkCreator(
447+
specs=[
448+
AppConfigFragmentCreatorSpec(
449+
config_name="theme",
450+
scope_type=AppConfigScopeType.PUBLIC,
451+
scope_id="public",
452+
config={"a": 1},
453+
),
454+
AppConfigFragmentCreatorSpec(
455+
config_name="theme",
456+
scope_type=AppConfigScopeType.DOMAIN,
457+
scope_id=_DOMAIN_ID,
458+
config={"b": 2},
459+
),
460+
]
461+
)
462+
)
463+
assert len(result.succeeded) == 2
464+
assert result.failed == []
465+
for fragment in result.succeeded:
466+
assert (await repository.get_by_id(fragment.id)).id == fragment.id
467+
468+
async def test_partial_when_one_not_allow_listed(
469+
self, repository: AppConfigFragmentRepository, menu_defined: None
470+
) -> None:
471+
result = await repository.bulk_create(
472+
BulkCreator(
473+
specs=[
474+
AppConfigFragmentCreatorSpec(
475+
config_name="theme", # allow-listed
476+
scope_type=AppConfigScopeType.DOMAIN,
477+
scope_id=_DOMAIN_ID,
478+
config={"a": 1},
479+
),
480+
AppConfigFragmentCreatorSpec(
481+
config_name="menu", # defined but NOT allow-listed -> FK rejects the insert
482+
scope_type=AppConfigScopeType.PUBLIC,
483+
scope_id="public",
484+
config={"b": 2},
485+
),
486+
]
487+
)
488+
)
489+
# partial: the allow-listed theme fragment is created; the menu item (index 1) is rejected
490+
assert [f.config_name for f in result.succeeded] == ["theme"]
491+
assert [f.index for f in result.failed] == [1]
492+
# the created theme fragment persists (not rolled back with the rejected one)
493+
search = await repository.admin_search(
494+
BatchQuerier(pagination=OffsetPagination(limit=10, offset=0))
495+
)
496+
assert {item.config_name for item in search.items} == {"theme"}
497+
498+
499+
class TestBulkUpdate:
500+
async def test_all_updated(
501+
self,
502+
repository: AppConfigFragmentRepository,
503+
two_fragments: list[AppConfigFragmentData],
504+
) -> None:
505+
result = await repository.bulk_update([
506+
Updater(
507+
spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"x": 1})),
508+
pk_value=two_fragments[0].id,
509+
),
510+
Updater(
511+
spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"y": 2})),
512+
pk_value=two_fragments[1].id,
513+
),
514+
])
515+
assert [u.config for u in result.succeeded] == [{"x": 1}, {"y": 2}]
516+
assert result.failed == []
517+
assert (await repository.get_by_id(two_fragments[0].id)).config == {"x": 1}
518+
519+
async def test_partial_when_one_missing(
520+
self,
521+
repository: AppConfigFragmentRepository,
522+
two_fragments: list[AppConfigFragmentData],
523+
) -> None:
524+
missing_id = AppConfigFragmentID(uuid.uuid4())
525+
result = await repository.bulk_update([
526+
Updater(
527+
spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"x": 1})),
528+
pk_value=two_fragments[0].id,
529+
),
530+
Updater(
531+
spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"z": 9})),
532+
pk_value=missing_id, # missing -> reported
533+
),
534+
])
535+
# partial: the existing fragment is updated; the missing one (index 1) is reported
536+
assert [u.config for u in result.succeeded] == [{"x": 1}]
537+
assert [f.index for f in result.failed] == [1]
538+
assert "not found" in result.failed[0].message
539+
assert (await repository.get_by_id(two_fragments[0].id)).config == {"x": 1}
540+
541+
542+
class TestBulkPurge:
543+
async def test_all_purged(
544+
self,
545+
repository: AppConfigFragmentRepository,
546+
two_fragments: list[AppConfigFragmentData],
547+
) -> None:
548+
result = await repository.bulk_purge([
549+
Purger(row_class=AppConfigFragmentRow, pk_value=fragment.id)
550+
for fragment in two_fragments
551+
])
552+
assert {p.id for p in result.succeeded} == {f.id for f in two_fragments}
553+
assert result.failed == []
554+
for fragment in two_fragments:
555+
with pytest.raises(AppConfigFragmentNotFound):
556+
await repository.get_by_id(fragment.id)
557+
558+
async def test_partial_when_one_missing(
559+
self,
560+
repository: AppConfigFragmentRepository,
561+
two_fragments: list[AppConfigFragmentData],
562+
) -> None:
563+
missing_id = AppConfigFragmentID(uuid.uuid4())
564+
result = await repository.bulk_purge([
565+
Purger(row_class=AppConfigFragmentRow, pk_value=two_fragments[0].id),
566+
Purger(row_class=AppConfigFragmentRow, pk_value=missing_id), # missing -> reported
567+
])
568+
# partial: the existing fragment is purged; the missing one (index 1) is reported
569+
assert [p.id for p in result.succeeded] == [two_fragments[0].id]
570+
assert [f.index for f in result.failed] == [1]
571+
assert "not found" in result.failed[0].message
572+
with pytest.raises(AppConfigFragmentNotFound):
573+
await repository.get_by_id(two_fragments[0].id)

0 commit comments

Comments
 (0)