feat: Implement System Model, Mock Source and Ingestion - #30
Merged
cschwartz merged 5 commits intoMar 14, 2026
Conversation
|
Warning Rate limit exceeded
⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughAdds CMDB ingestion end-to-end: new System model and association table, CMDB mock and generic mock factory, CMDB ingestion service and routes (including POST /ingest and /systems), refactors assets to public models, updates queries/schemas, Alembic migration, tests, and pattern docs. Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client/User
participant Backend as Verdict Backend
participant AssetMock as Asset Mock
participant CMDBMock as CMDB Mock
participant DB as Database
Client->>Backend: POST /ingest
activate Backend
rect rgba(100, 200, 150, 0.5)
Note over Backend,AssetMock: Asset ingestion phase
Backend->>AssetMock: GET /assets
AssetMock-->>Backend: [AssetIndexItem,...]
loop per asset
Backend->>AssetMock: GET /assets/{id}
AssetMock-->>Backend: AssetDetail
Backend->>Backend: validate & convert -> AssetCreate
end
Backend->>DB: upsert assets by gold_source
DB-->>Backend: list[AssetPublic]
end
rect rgba(150, 150, 200, 0.5)
Note over Backend,CMDBMock: CMDB ingestion phase
Backend->>CMDBMock: GET /systems
CMDBMock-->>Backend: [SystemIndexItem,...]
loop per system
Backend->>CMDBMock: GET /systems/{id}
CMDBMock-->>Backend: SystemDetail
Backend->>Backend: resolve asset_gold_source_ids -> asset IDs
Backend->>Backend: validate & convert -> SystemCreate
end
Backend->>DB: upsert systems by gold_source
DB-->>Backend: list[SystemPublic]
Backend->>DB: sync asset_system links (diff)
end
Backend->>DB: commit
DB-->>Backend: commit ok
Backend-->>Client: FullIngestionResponse {assets_ingested, systems_ingested}
deactivate Backend
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
📝 Coding Plan
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (6)
verdict-backend/alembic/versions/3bfa8f122226_create_system_and_asset_system_tables.py (2)
46-53: Consider adding an index onsystem_idfor reverse lookups.The composite primary key
(asset_id, system_id)efficiently supports queries byasset_id, but queries that filter bysystem_id(e.g., "find all assets for a system") may benefit from an additional index.💡 Optional: Add index for system_id lookups
sa.ForeignKeyConstraint(["asset_id"], ["asset.id"]), sa.ForeignKeyConstraint(["system_id"], ["system.id"]), sa.PrimaryKeyConstraint("asset_id", "system_id"), ) + op.create_index("ix_asset_system_system_id", "asset_system", ["system_id"])And in downgrade:
def downgrade() -> None: """Downgrade schema.""" + op.drop_index("ix_asset_system_system_id", table_name="asset_system") op.drop_table("asset_system") op.drop_table("system")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/alembic/versions/3bfa8f122226_create_system_and_asset_system_tables.py` around lines 46 - 53, Add a non-unique index on the asset_system table's system_id to speed reverse lookups: after the op.create_table call that defines asset_system (and its columns asset_id and system_id) add an op.create_index for system_id (name it something like ix_asset_system_system_id) and in the downgrade add the corresponding op.drop_index before dropping the asset_system table to keep migrations reversible.
50-51: Consider specifying ON DELETE behavior for foreign keys.The foreign key constraints don't specify cascade behavior. If an asset or system is deleted, the join table rows will become orphaned or cause constraint violations depending on database defaults.
♻️ Proposed fix to add explicit cascade behavior
- sa.ForeignKeyConstraint(["asset_id"], ["asset.id"]), - sa.ForeignKeyConstraint(["system_id"], ["system.id"]), + sa.ForeignKeyConstraint(["asset_id"], ["asset.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["system_id"], ["system.id"], ondelete="CASCADE"),🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/alembic/versions/3bfa8f122226_create_system_and_asset_system_tables.py` around lines 50 - 51, The migration's ForeignKeyConstraint definitions for asset_id -> asset.id and system_id -> system.id should include explicit ON DELETE behavior to avoid orphaned join rows; update the sa.ForeignKeyConstraint entries (the ones referencing "asset_id" -> "asset.id" and "system_id" -> "system.id") to specify ondelete="CASCADE" (or another desired policy) so deletes of Asset or System cascade to the join table rows.docs/patterns/ingestion.md (1)
17-20: Document dependency order in the orchestrator section.Line 19's reference-resolution rule only works if the orchestrator executes sources in dependency order, e.g. assets before systems. Making that explicit here will save future sources from following the pattern and then failing deterministically on missing references.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/patterns/ingestion.md` around lines 17 - 20, Clarify that the global ingestion endpoint (POST /ingest) and the orchestrator must execute source pipelines in dependency order so reference-resolution by gold source ID works reliably; update the orchestrator section to state that the orchestrator determines and enforces dependency order (e.g., ingest assets before systems) and will schedule/sequence sources accordingly so that references can be resolved during each pipeline run and the session commit/rollback semantics remain intact.verdict-backend/app/models/system.py (1)
35-36: UseField(default_factory=list)for the mutable default.While Pydantic handles mutable defaults safely in most cases, using
default_factoryis explicit and avoids any potential confusion or edge cases.♻️ Suggested fix
-class SystemPublic(SystemBase, PublicModel): - asset_ids: list[int] = [] +class SystemPublic(SystemBase, PublicModel): + asset_ids: list[int] = Field(default_factory=list)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/app/models/system.py` around lines 35 - 36, The mutable default for asset_ids in class SystemPublic should use Pydantic's default_factory: update the SystemPublic model (symbol: SystemPublic, attribute: asset_ids) to declare asset_ids with Field(default_factory=list) instead of assigning [] directly, and add the necessary import of Field from pydantic so the model uses a fresh list per instance.verdict-backend/app/services/cmdb_ingestion.py (1)
75-99: Consider batching asset resolution for better performance at scale.The current implementation performs one DB query per gold source ID. While acceptable for the current synchronous ingestion model, this could become a bottleneck with large datasets.
♻️ Potential optimization using a single query
def _resolve_asset_ids( session: Session, gold_source_ids: list[str], ) -> Result[list[int], ValidationError | DBError]: if not gold_source_ids: return Ok([]) try: assets = session.exec( select(Asset).where( Asset.gold_source_type == GoldSourceType.ASSET_INVENTORY, Asset.gold_source_id.in_(gold_source_ids), ) ).all() except OperationalError as e: return Err(db_error_from(e)) found_ids = {a.gold_source_id: a.id for a in assets} missing = set(gold_source_ids) - found_ids.keys() if missing: return Err(ValidationError(raw=f"unresolvable asset references: {missing}")) return Ok([found_ids[gs_id] for gs_id in gold_source_ids])🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/app/services/cmdb_ingestion.py` around lines 75 - 99, The loop in _resolve_asset_ids issues one DB query per gold_source_id; replace it with a single batched query using session.exec(select(Asset).where(Asset.gold_source_type == GoldSourceType.ASSET_INVENTORY, Asset.gold_source_id.in_(gold_source_ids))) to fetch all matching assets at once, catch DB exceptions and convert them to the existing DBError (e.g., via db_error_from), build a mapping from asset.gold_source_id to asset.id, detect any missing gold_source_ids and return a ValidationError listing them, and finally return Ok of ids in the same order as the input gold_source_ids; also handle the empty gold_source_ids case by returning Ok([]) immediately.verdict-backend/app/services/asset_ingestion.py (1)
95-99: Consider removing redundant flush.Since
_upsert_assetnow flushes after each operation (lines 122 and 130), this final flush at line 96 is effectively a no-op. All database changes have already been flushed by the time the loop completes.♻️ Proposed simplification
assets.append(upsert_result.value) - try: - session.flush() - except OperationalError as e: - return Err(db_error_from(e)) return Ok(assets)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/app/services/asset_ingestion.py` around lines 95 - 99, The final try/except around session.flush() is redundant because _upsert_asset already flushes per operation (see calls inside _upsert_asset at lines where it flushes after insert/update), so remove the outer try/except and the session.flush() call in asset_ingestion.py and leave the function returning Ok(assets); ensure any error handling remains via the existing db_error_from handling inside _upsert_asset rather than duplicating the flush/error block.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@verdict-backend/app/services/cmdb_ingestion.py`:
- Around line 135-157: The _sync_asset_links function currently performs
session.execute calls for sa.delete(asset_system) and sa.insert(asset_system)
without handling DB errors; wrap the delete and insert operations (the
session.execute calls that use sa.delete(asset_system) and
sa.insert(asset_system)) in a try/except that catches
sqlalchemy.exc.OperationalError (or the DB OperationalError alias you use) and
on exception log the error and return False, otherwise return True after all ops
succeed; update the caller to check the boolean result from _sync_asset_links
and handle failures accordingly.
In `@verdict-backend/devenv.nix`:
- Around line 161-180: The curl readiness probes using the patterns "until curl
-sf http://localhost:4010/assets" and "until curl -sf
http://localhost:4011/systems" need an explicit timeout so the retry loop can
progress; update both curl invocations to include a short per-request timeout
option (e.g., --max-time and/or --connect-timeout) so each probe fails fast and
the existing retries variable/30-retry logic can trigger, and apply the same
timeout change to the similar verdict app probe mentioned later so all readiness
checks are bounded.
In `@verdict-backend/tests/routes/test_ingestion.py`:
- Around line 47-60: The test sets app.dependency_overrides[get_http_client] =
_mock_http_client but does not guarantee cleanup on failures; wrap the request
and assertions in a try/finally (or move the override into a fixture) so that
app.dependency_overrides.pop(get_http_client, None) always runs; locate the
override assignment and ensure cleanup for get_http_client (and similarly for
the other occurrences around lines 80-90) by placing the pop(...) in the finally
block so the override is removed even if app_client.post("/ingest") or
subsequent asserts fail.
- Around line 64-88: The test test_full_ingest_rolls_back_on_system_failure
currently only asserts a 502 response but doesn't verify that the DB rollback
occurred; after awaiting response = await app_client.post("/ingest") use the
provided db_session to query the relevant tables/entities (e.g., assets and
systems created by the ingest flow, referencing ASSET_INDEX/ASSET_DETAIL and
SYSTEM_INDEX) and assert that no new rows were inserted (counts are zero or
unchanged), or alternatively assert specific entities (by IDs from
ASSET_INDEX/ASSET_DETAIL) are absent; update the test to fail if any of those
persisted so it guarantees the transaction was rolled back.
---
Nitpick comments:
In `@docs/patterns/ingestion.md`:
- Around line 17-20: Clarify that the global ingestion endpoint (POST /ingest)
and the orchestrator must execute source pipelines in dependency order so
reference-resolution by gold source ID works reliably; update the orchestrator
section to state that the orchestrator determines and enforces dependency order
(e.g., ingest assets before systems) and will schedule/sequence sources
accordingly so that references can be resolved during each pipeline run and the
session commit/rollback semantics remain intact.
In
`@verdict-backend/alembic/versions/3bfa8f122226_create_system_and_asset_system_tables.py`:
- Around line 46-53: Add a non-unique index on the asset_system table's
system_id to speed reverse lookups: after the op.create_table call that defines
asset_system (and its columns asset_id and system_id) add an op.create_index for
system_id (name it something like ix_asset_system_system_id) and in the
downgrade add the corresponding op.drop_index before dropping the asset_system
table to keep migrations reversible.
- Around line 50-51: The migration's ForeignKeyConstraint definitions for
asset_id -> asset.id and system_id -> system.id should include explicit ON
DELETE behavior to avoid orphaned join rows; update the sa.ForeignKeyConstraint
entries (the ones referencing "asset_id" -> "asset.id" and "system_id" ->
"system.id") to specify ondelete="CASCADE" (or another desired policy) so
deletes of Asset or System cascade to the join table rows.
In `@verdict-backend/app/models/system.py`:
- Around line 35-36: The mutable default for asset_ids in class SystemPublic
should use Pydantic's default_factory: update the SystemPublic model (symbol:
SystemPublic, attribute: asset_ids) to declare asset_ids with
Field(default_factory=list) instead of assigning [] directly, and add the
necessary import of Field from pydantic so the model uses a fresh list per
instance.
In `@verdict-backend/app/services/asset_ingestion.py`:
- Around line 95-99: The final try/except around session.flush() is redundant
because _upsert_asset already flushes per operation (see calls inside
_upsert_asset at lines where it flushes after insert/update), so remove the
outer try/except and the session.flush() call in asset_ingestion.py and leave
the function returning Ok(assets); ensure any error handling remains via the
existing db_error_from handling inside _upsert_asset rather than duplicating the
flush/error block.
In `@verdict-backend/app/services/cmdb_ingestion.py`:
- Around line 75-99: The loop in _resolve_asset_ids issues one DB query per
gold_source_id; replace it with a single batched query using
session.exec(select(Asset).where(Asset.gold_source_type ==
GoldSourceType.ASSET_INVENTORY, Asset.gold_source_id.in_(gold_source_ids))) to
fetch all matching assets at once, catch DB exceptions and convert them to the
existing DBError (e.g., via db_error_from), build a mapping from
asset.gold_source_id to asset.id, detect any missing gold_source_ids and return
a ValidationError listing them, and finally return Ok of ids in the same order
as the input gold_source_ids; also handle the empty gold_source_ids case by
returning Ok([]) immediately.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 4e53066c-3f75-4625-ae24-8b725dd8e6a8
⛔ Files ignored due to path filters (2)
devenv.lockis excluded by!**/*.lockverdict-backend/devenv.lockis excluded by!**/*.lock
📒 Files selected for processing (47)
devenv.yamldocs/patterns/backend.mddocs/patterns/error-handling.mddocs/patterns/ingestion.mddocs/patterns/mock-services.mddocs/patterns/models.mddocs/patterns/routes.mdmock-services/asset_inventory/app.pymock-services/cmdb/__init__.pymock-services/cmdb/app.pymock-services/cmdb/data/systems.yamlmock-services/cmdb/models.pymock-services/devenv.nixmock-services/justfilemock-services/mock_helpers.pyverdict-backend/alembic/versions/3bfa8f122226_create_system_and_asset_system_tables.pyverdict-backend/app/config.pyverdict-backend/app/main.pyverdict-backend/app/models/__init__.pyverdict-backend/app/models/asset.pyverdict-backend/app/models/base.pyverdict-backend/app/models/base_model.pyverdict-backend/app/models/gold_source.pyverdict-backend/app/models/system.pyverdict-backend/app/models/timestamp.pyverdict-backend/app/queries.pyverdict-backend/app/routes/assets.pyverdict-backend/app/routes/ingestion.pyverdict-backend/app/routes/systems.pyverdict-backend/app/schemas/asset.pyverdict-backend/app/schemas/base.pyverdict-backend/app/schemas/external/cmdb.pyverdict-backend/app/schemas/ingestion.pyverdict-backend/app/schemas/system.pyverdict-backend/app/services/asset_ingestion.pyverdict-backend/app/services/cmdb_ingestion.pyverdict-backend/devenv.nixverdict-backend/devenv.yamlverdict-backend/tests/e2e/test_asset_ingestion.pyverdict-backend/tests/e2e/test_cmdb_ingestion.pyverdict-backend/tests/factories/system.pyverdict-backend/tests/models/mixin_test_model.pyverdict-backend/tests/routes/test_assets.pyverdict-backend/tests/routes/test_ingestion.pyverdict-backend/tests/routes/test_systems.pyverdict-backend/tests/services/test_asset_ingestion.pyverdict-backend/tests/services/test_cmdb_ingestion.py
💤 Files with no reviewable changes (4)
- docs/patterns/backend.md
- verdict-backend/app/schemas/base.py
- verdict-backend/app/models/timestamp.py
- verdict-backend/tests/routes/test_assets.py
Comment on lines
161
to
+180
| until curl -sf http://localhost:4010/assets > /dev/null 2>&1; do | ||
| retries=$((retries + 1)) | ||
| if [ $retries -ge 30 ]; then | ||
| echo "ERROR: Mock service failed to start" | ||
| echo "ERROR: Asset inventory mock failed to start" | ||
| exit 1 | ||
| fi | ||
| sleep 1 | ||
| done | ||
| echo "Asset inventory mock ready" | ||
|
|
||
| retries=0 | ||
| until curl -sf http://localhost:4011/systems > /dev/null 2>&1; do | ||
| retries=$((retries + 1)) | ||
| if [ $retries -ge 30 ]; then | ||
| echo "ERROR: CMDB mock failed to start" | ||
| exit 1 | ||
| fi | ||
| sleep 1 | ||
| done | ||
| echo "Mock service ready" | ||
| echo "CMDB mock ready" |
There was a problem hiding this comment.
Add timeouts to the mock readiness curls.
curl -sf has no response timeout, so a half-started local service can hang a probe indefinitely and the 30-retry cap never fires. Please bound these calls, then mirror the same change in the verdict app probe below.
⏱️ Proposed fix
- until curl -sf http://localhost:4010/assets > /dev/null 2>&1; do
+ until curl -sf --connect-timeout 1 --max-time 2 http://localhost:4010/assets > /dev/null 2>&1; do
- until curl -sf http://localhost:4011/systems > /dev/null 2>&1; do
+ until curl -sf --connect-timeout 1 --max-time 2 http://localhost:4011/systems > /dev/null 2>&1; do📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| until curl -sf http://localhost:4010/assets > /dev/null 2>&1; do | |
| retries=$((retries + 1)) | |
| if [ $retries -ge 30 ]; then | |
| echo "ERROR: Mock service failed to start" | |
| echo "ERROR: Asset inventory mock failed to start" | |
| exit 1 | |
| fi | |
| sleep 1 | |
| done | |
| echo "Asset inventory mock ready" | |
| retries=0 | |
| until curl -sf http://localhost:4011/systems > /dev/null 2>&1; do | |
| retries=$((retries + 1)) | |
| if [ $retries -ge 30 ]; then | |
| echo "ERROR: CMDB mock failed to start" | |
| exit 1 | |
| fi | |
| sleep 1 | |
| done | |
| echo "Mock service ready" | |
| echo "CMDB mock ready" | |
| until curl -sf --connect-timeout 1 --max-time 2 http://localhost:4010/assets > /dev/null 2>&1; do | |
| retries=$((retries + 1)) | |
| if [ $retries -ge 30 ]; then | |
| echo "ERROR: Asset inventory mock failed to start" | |
| exit 1 | |
| fi | |
| sleep 1 | |
| done | |
| echo "Asset inventory mock ready" | |
| retries=0 | |
| until curl -sf --connect-timeout 1 --max-time 2 http://localhost:4011/systems > /dev/null 2>&1; do | |
| retries=$((retries + 1)) | |
| if [ $retries -ge 30 ]; then | |
| echo "ERROR: CMDB mock failed to start" | |
| exit 1 | |
| fi | |
| sleep 1 | |
| done | |
| echo "CMDB mock ready" |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@verdict-backend/devenv.nix` around lines 161 - 180, The curl readiness probes
using the patterns "until curl -sf http://localhost:4010/assets" and "until curl
-sf http://localhost:4011/systems" need an explicit timeout so the retry loop
can progress; update both curl invocations to include a short per-request
timeout option (e.g., --max-time and/or --connect-timeout) so each probe fails
fast and the existing retries variable/30-retry logic can trigger, and apply the
same timeout change to the similar verdict app probe mentioned later so all
readiness checks are bounded.
Owner
Author
There was a problem hiding this comment.
Correct, the test harness seems brittle. We'll do something smarter here. Just not in this ticket.
There was a problem hiding this comment.
@cschwartz, understood — happy to defer this to a future ticket. Would you like me to open a GitHub issue to track the test harness improvements (bounded curl timeouts and any other hardening you have in mind)?
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
verdict-backend/tests/e2e/conftest.py (1)
14-19: Isolate DB state per test instead of per session.At Line 14,
scope="session"means data created by one test can leak into subsequent tests until the whole session ends. Prefer function-scoped cleanup for test independence.Proposed change
-@pytest.fixture(scope="session", autouse=True) +@pytest.fixture(autouse=True) def _clean_db() -> Generator[None, None, None]: """Truncate all tables before and after the e2e session.""" _truncate() yield _truncate()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/tests/e2e/conftest.py` around lines 14 - 19, The _clean_db fixture currently uses scope="session" which allows DB state to persist across tests; change its scope to "function" (or remove scope to use default function scope) so that _truncate() runs before and after each individual test, ensuring isolation; locate the _clean_db fixture and adjust its scope attribute while keeping the existing calls to _truncate() in the setup and teardown logic (referencing _clean_db and the _truncate() helper).verdict-backend/tests/routes/test_ingestion.py (1)
58-67: Consider using or removingdb_session.The
db_sessionfixture is injected but unused. If it's solely for test isolation, that's fine—but adding a DB assertion (e.g., verifying one Asset and one System were persisted) would strengthen the test.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/tests/routes/test_ingestion.py` around lines 58 - 67, The test_full_ingest_endpoint currently accepts the db_session fixture but never uses it; either remove db_session from the test signature or add assertions that verify persistence (e.g., query the DB via db_session to assert one Asset and one System exist after calling app_client.post("/ingest")). Locate test_full_ingest_endpoint and either drop the db_session parameter or use db_session to query the Asset and System models (or repository methods) to assert counts/records match the expected assets_ingested and systems_ingested.verdict-backend/app/services/cmdb_ingestion.py (2)
75-99: Resolve asset IDs fromAssetin bulk here.This helper does one lookup and one
AssetPublicvalidation per reference, andingest_systems()repeats that for every system. A singleSELECT Asset.gold_source_id, Asset.id ... IN (...)for the unique gold-source IDs would remove the N+1 query pattern and keep this path decoupled from API-model validation.Also applies to: 180-186
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/app/services/cmdb_ingestion.py` around lines 75 - 99, The _resolve_asset_ids helper currently does N+1 lookups and API-model validation per gold_source_id; replace that with a single bulk query that selects Asset.gold_source_id and Asset.id WHERE Asset.gold_source_id IN (unique gold_source_ids) using the Session (referencing _resolve_asset_ids, GoldSourceMixin.get_by_gold_source should not be used here), build a map from gold_source_id to id, and then iterate the original gold_source_ids to produce asset_ids in the same order, returning a ValidationError via Err if any gold_source_id is missing; apply the same bulk-query pattern to the analogous logic referenced at lines ~180-186.
102-132: DelaySystemPublicvalidation until after link sync.
_upsert_system()only needs to hand the caller something with anid, but it validatesSystemPublicbefore_sync_asset_links()runs. ReturningSystemorsystem_idhere and buildingSystemPubliconce after the asset links are synchronized keeps the write path cleaner and avoids returning pre-link state.Also applies to: 193-200
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@verdict-backend/app/services/cmdb_ingestion.py` around lines 102 - 132, _upsert_system currently validates and returns SystemPublic before asset links are synchronized; change it to return the persisted System instance (or its id) instead of calling SystemPublic.model_validate so that validation is deferred until after _sync_asset_links runs. Concretely, in _upsert_system replace the Ok(SystemPublic.model_validate(...)) returns with Ok(existing) or Ok(new) (or Ok(existing.id)/Ok(new.id) if the caller only needs an id), remove/avoid SystemPublic.model_validate calls here, and update the corresponding call sites (including the similar block around the 193-200 range) so they call SystemPublic.model_validate only after _sync_asset_links has completed.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@verdict-backend/app/services/cmdb_ingestion.py`:
- Around line 140-142: The initial select against asset_system in
_sync_asset_links (the
session.execute(sa.select(asset_system.c.asset_id).where(asset_system.c.system_id
== system_id)).all() call) must be wrapped in the same OperationalError/DBError
handling as the subsequent writes: catch sqlalchemy.exc.OperationalError (or the
project's DBError translation), convert it to the Result[None, DBError] failure
and return early instead of letting the raw exception escape; apply the same
pattern to the later read blocks around lines 149-160 so all DB reads in
_sync_asset_links return the unified Result[None, DBError] on failure.
In `@verdict-backend/tests/routes/test_ingestion.py`:
- Around line 98-107: The test
test_full_ingest_does_not_persist_on_system_failure currently only asserts that
System rows are rolled back; add a parallel assertion that no Asset rows were
persisted to verify transaction atomicity. After the existing systems query
(db_session.exec(select(System)).all()), query assets with
db_session.exec(select(Asset)).all() and assert its length is 0 (or assert not
assets) so the test ensures both System and Asset records are rolled back when
_mock_system_failure() is triggered during app_client.post("/ingest").
---
Nitpick comments:
In `@verdict-backend/app/services/cmdb_ingestion.py`:
- Around line 75-99: The _resolve_asset_ids helper currently does N+1 lookups
and API-model validation per gold_source_id; replace that with a single bulk
query that selects Asset.gold_source_id and Asset.id WHERE Asset.gold_source_id
IN (unique gold_source_ids) using the Session (referencing _resolve_asset_ids,
GoldSourceMixin.get_by_gold_source should not be used here), build a map from
gold_source_id to id, and then iterate the original gold_source_ids to produce
asset_ids in the same order, returning a ValidationError via Err if any
gold_source_id is missing; apply the same bulk-query pattern to the analogous
logic referenced at lines ~180-186.
- Around line 102-132: _upsert_system currently validates and returns
SystemPublic before asset links are synchronized; change it to return the
persisted System instance (or its id) instead of calling
SystemPublic.model_validate so that validation is deferred until after
_sync_asset_links runs. Concretely, in _upsert_system replace the
Ok(SystemPublic.model_validate(...)) returns with Ok(existing) or Ok(new) (or
Ok(existing.id)/Ok(new.id) if the caller only needs an id), remove/avoid
SystemPublic.model_validate calls here, and update the corresponding call sites
(including the similar block around the 193-200 range) so they call
SystemPublic.model_validate only after _sync_asset_links has completed.
In `@verdict-backend/tests/e2e/conftest.py`:
- Around line 14-19: The _clean_db fixture currently uses scope="session" which
allows DB state to persist across tests; change its scope to "function" (or
remove scope to use default function scope) so that _truncate() runs before and
after each individual test, ensuring isolation; locate the _clean_db fixture and
adjust its scope attribute while keeping the existing calls to _truncate() in
the setup and teardown logic (referencing _clean_db and the _truncate() helper).
In `@verdict-backend/tests/routes/test_ingestion.py`:
- Around line 58-67: The test_full_ingest_endpoint currently accepts the
db_session fixture but never uses it; either remove db_session from the test
signature or add assertions that verify persistence (e.g., query the DB via
db_session to assert one Asset and one System exist after calling
app_client.post("/ingest")). Locate test_full_ingest_endpoint and either drop
the db_session parameter or use db_session to query the Asset and System models
(or repository methods) to assert counts/records match the expected
assets_ingested and systems_ingested.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3c83454d-b077-45ef-8e03-f0c376922268
⛔ Files ignored due to path filters (1)
mock-services/devenv.lockis excluded by!**/*.lock
📒 Files selected for processing (7)
.github/workflows/verdict-backend.yaml.gitignoreverdict-backend/app/services/cmdb_ingestion.pyverdict-backend/devenv.nixverdict-backend/justfileverdict-backend/tests/e2e/conftest.pyverdict-backend/tests/routes/test_ingestion.py
✅ Files skipped from review due to trivial changes (1)
- .gitignore
🚧 Files skipped from review as they are similar to previous changes (1)
- verdict-backend/devenv.nix
Comment on lines
+140
to
+142
| existing_rows = session.execute( | ||
| sa.select(asset_system.c.asset_id).where(asset_system.c.system_id == system_id) | ||
| ).all() |
There was a problem hiding this comment.
Handle read-path DB failures in _sync_asset_links().
The initial lookup is still outside the OperationalError guard, so a failing read here bypasses the Result[None, DBError] contract and leaks a raw DB exception out of ingestion.
🛡️ Proposed fix
def _sync_asset_links(
session: Session,
system_id: int,
desired_asset_ids: list[int],
) -> Result[None, DBError]:
- existing_rows = session.execute(
- sa.select(asset_system.c.asset_id).where(asset_system.c.system_id == system_id)
- ).all()
+ try:
+ existing_rows = session.execute(
+ sa.select(asset_system.c.asset_id).where(asset_system.c.system_id == system_id)
+ ).all()
+ except OperationalError as e:
+ return Err(db_error_from(e))
existing_ids = {row.asset_id for row in existing_rows}
desired_ids = set(desired_asset_ids)
to_remove = existing_ids - desired_ids
to_add = desired_ids - existing_idsAlso applies to: 149-160
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@verdict-backend/app/services/cmdb_ingestion.py` around lines 140 - 142, The
initial select against asset_system in _sync_asset_links (the
session.execute(sa.select(asset_system.c.asset_id).where(asset_system.c.system_id
== system_id)).all() call) must be wrapped in the same OperationalError/DBError
handling as the subsequent writes: catch sqlalchemy.exc.OperationalError (or the
project's DBError translation), convert it to the Result[None, DBError] failure
and return early instead of letting the raw exception escape; apply the same
pattern to the later read blocks around lines 149-160 so all DB reads in
_sync_asset_links return the unified Result[None, DBError] on failure.
Comment on lines
+98
to
+107
| @respx.mock | ||
| async def test_full_ingest_does_not_persist_on_system_failure( | ||
| app_client: AsyncClient, db_session: Session, mock_http_client | ||
| ): | ||
| _mock_system_failure() | ||
|
|
||
| await app_client.post("/ingest") | ||
|
|
||
| systems = db_session.exec(select(System)).all() | ||
| assert len(systems) == 0 |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify Asset model exists and can be imported
ast-grep --pattern 'class Asset'Repository: cschwartz/verdict
Length of output: 792
🏁 Script executed:
# First, let's see the full test file and endpoint implementation
cat -n verdict-backend/tests/routes/test_ingestion.py | head -120Repository: cschwartz/verdict
Length of output: 3935
🏁 Script executed:
# Find the ingest endpoint
find verdict-backend -name "*.py" -type f | xargs rg -l "POST.*ingest|@.*\.post.*ingest" | head -5Repository: cschwartz/verdict
Length of output: 100
🏁 Script executed:
# Search for the ingest route definition
rg -A 20 "async def.*ingest|@.*route.*ingest" verdict-backend/app/routes/ -t pyRepository: cschwartz/verdict
Length of output: 1384
🏁 Script executed:
# Get full ingestion.py to understand transaction handling
cat -n verdict-backend/app/routes/ingestion.pyRepository: cschwartz/verdict
Length of output: 1324
🏁 Script executed:
# Find unwrap_or_raise implementation
rg -B 5 -A 10 "def unwrap_or_raise" verdict-backend/ -t pyRepository: cschwartz/verdict
Length of output: 1022
🏁 Script executed:
# Check ingest_systems to see if it validates asset references
rg -B 5 -A 30 "def ingest_systems" verdict-backend/ -t pyRepository: cschwartz/verdict
Length of output: 3006
🏁 Script executed:
# Check how get_session handles exceptions and rollback
cat -n verdict-backend/app/db.pyRepository: cschwartz/verdict
Length of output: 695
🏁 Script executed:
# Verify the asset ingestion process writes to database
rg -B 5 -A 30 "def ingest_assets" verdict-backend/app/services/ -t pyRepository: cschwartz/verdict
Length of output: 2813
🏁 Script executed:
# Confirm _resolve_asset_ids validates asset existence
rg -B 2 -A 10 "_resolve_asset_ids" verdict-backend/app/services/cmdb_ingestion.py -t pyRepository: cschwartz/verdict
Length of output: 956
Add assertion verifying Asset rollback.
The test validates that System records don't persist on system failure, but assets are ingested and committed to the same transaction first. To fully verify the atomicity guarantee, also assert that no Asset records persist.
💚 Proposed enhancement
from app.models.system import System
+from app.models.asset import Asset
`@respx.mock`
async def test_full_ingest_does_not_persist_on_system_failure(
app_client: AsyncClient, db_session: Session, mock_http_client
):
_mock_system_failure()
await app_client.post("/ingest")
systems = db_session.exec(select(System)).all()
assert len(systems) == 0
+ assets = db_session.exec(select(Asset)).all()
+ assert len(assets) == 0🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@verdict-backend/tests/routes/test_ingestion.py` around lines 98 - 107, The
test test_full_ingest_does_not_persist_on_system_failure currently only asserts
that System rows are rolled back; add a parallel assertion that no Asset rows
were persisted to verify transaction atomicity. After the existing systems query
(db_session.exec(select(System)).all()), query assets with
db_session.exec(select(Asset)).all() and assert its length is 0 (or assert not
assets) so the test ensures both System and Asset records are rolled back when
_mock_system_failure() is triggered during app_client.post("/ingest").
Owner
Author
There was a problem hiding this comment.
this is tested in another test, we only do one assertion (but maybe with multiple assert statements) per test^
There was a problem hiding this comment.
@cschwartz, understood — one logical concept per test, with atomicity of Asset rollback covered elsewhere. I'll keep this in mind for future reviews.
✏️ Learnings added
Learnt from: cschwartz
Repo: cschwartz/verdict PR: 30
File: verdict-backend/tests/routes/test_ingestion.py:98-107
Timestamp: 2026-03-14T09:28:07.512Z
Learning: In cschwartz/verdict, the testing convention is one logical assertion (one concept) per test, even if that means multiple assert statements. Separate concerns (e.g., Asset rollback vs. System rollback) belong in separate tests rather than being combined into a single test.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
🧠 Learnings used
Learnt from: cschwartz
Repo: cschwartz/verdict PR: 26
File: verdict-backend/app/services/asset_ingestion.py:111-129
Timestamp: 2026-03-01T07:49:07.243Z
Learning: In verdict-backend, asset ingestion runs synchronously via a single POST endpoint at /assets/ingest, with no concurrent ingestion. Database transactions serialize requests, so TOCTOU race conditions in the upsert logic (get_by_gold_source → session.add) are not a current concern. Conflict handling (INSERT...ON CONFLICT or retry logic) is deferred until background workers or parallel ingestion are introduced.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@verdict-backend/devenv.nix`:
- Line 153: The pytest invocation that writes a JUnit XML (the line starting
with "DATABASE_NAME=${database_name}_test uv run pytest
--disable-plugin-autoload -p asyncio -m 'not e2e'
--junit-xml=test-results/unit.xml") assumes test-results/ exists; modify the
build/script to ensure the test-results directory is created (e.g., create the
directory if missing) before running that pytest command, and apply the same
change to the other occurrence of this pytest invocation elsewhere in the file.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 4896847b-5fa6-4aa5-ac93-f10e188d2af3
📒 Files selected for processing (2)
.github/workflows/verdict-backend.yamlverdict-backend/devenv.nix
| DATABASE_NAME=${database_name}_test just db-migrate | ||
| just check | ||
| DATABASE_NAME=${database_name}_test uv run pytest --disable-plugin-autoload -p asyncio -m 'not e2e' | ||
| DATABASE_NAME=${database_name}_test uv run pytest --disable-plugin-autoload -p asyncio -m 'not e2e' --junit-xml=test-results/unit.xml |
There was a problem hiding this comment.
Create test-results/ before writing JUnit XML.
The current commands assume test-results already exists. If it doesn’t, XML report creation can fail and downstream reporting breaks.
🛠️ Proposed fix
just db-test-reset
DATABASE_NAME=${database_name}_test just db-migrate
just check
+ mkdir -p test-results
DATABASE_NAME=${database_name}_test uv run pytest --disable-plugin-autoload -p asyncio -m 'not e2e' --junit-xml=test-results/unit.xml
@@
just db-test-reset
DATABASE_NAME=${database_name}_test just db-migrate
DATABASE_NAME=${database_name}_test uv run pytest --disable-plugin-autoload -p asyncio -m e2e --junit-xml=test-results/e2e.xmlAlso applies to: 218-218
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@verdict-backend/devenv.nix` at line 153, The pytest invocation that writes a
JUnit XML (the line starting with "DATABASE_NAME=${database_name}_test uv run
pytest --disable-plugin-autoload -p asyncio -m 'not e2e'
--junit-xml=test-results/unit.xml") assumes test-results/ exists; modify the
build/script to ensure the test-results directory is created (e.g., create the
directory if missing) before running that pytest command, and apply the same
change to the other occurrence of this pytest invocation elsewhere in the file.
Owner
Author
|
Build fail due to failed psycopg build in devenv-nixpkgs/rolling |
cschwartz
deleted the
4-implement-system-model-mock-source-and-ingestion
branch
March 14, 2026 10:11
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implements #4
Summary by CodeRabbit
New Features
Documentation
Tests