Skip to content

Commit 801351d

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 801351d

5 files changed

Lines changed: 285 additions & 0 deletions

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

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

Lines changed: 22 additions & 0 deletions
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
)
@@ -75,3 +76,24 @@ async def scoped_search(
7576
self, querier: BatchQuerier, scopes: Sequence[SearchScope]
7677
) -> AppConfigFragmentSearchResult:
7778
return await self._db_source.scoped_search(querier, scopes)
79+
80+
@app_config_fragment_repository_resilience.apply()
81+
async def bulk_create(
82+
self,
83+
creators: Sequence[Creator[AppConfigFragmentRow]],
84+
) -> AppConfigFragmentBulkWriteResult:
85+
return await self._db_source.bulk_create(creators)
86+
87+
@app_config_fragment_repository_resilience.apply()
88+
async def bulk_update(
89+
self,
90+
updaters: Sequence[Updater[AppConfigFragmentRow]],
91+
) -> AppConfigFragmentBulkWriteResult:
92+
return await self._db_source.bulk_update(updaters)
93+
94+
@app_config_fragment_repository_resilience.apply()
95+
async def bulk_purge(
96+
self,
97+
purgers: Sequence[Purger[AppConfigFragmentRow]],
98+
) -> AppConfigFragmentBulkWriteResult:
99+
return await self._db_source.bulk_purge(purgers)

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

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

0 commit comments

Comments
 (0)