feat: validate datastore write records - #16
Open
luccasmmg wants to merge 1 commit into
Open
Conversation
📝 WalkthroughWalkthroughChangesDatastore write validation
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant WriteService
participant Frictionless
participant CKAN
participant DatastoreEngine
Client->>WriteService: create_datastore(records, schema)
WriteService->>Frictionless: validate records
alt records invalid
Frictionless-->>WriteService: validation report
WriteService-->>Client: ValidationError
else records valid
WriteService->>CKAN: create resource
end
Client->>WriteService: upsert_datastore(records)
WriteService->>CKAN: load stored resource schema
WriteService->>Frictionless: validate records
alt records invalid
Frictionless-->>WriteService: validation report
WriteService-->>Client: ValidationError
else records valid
WriteService->>DatastoreEngine: write records
end
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@datastore/services/write.py`:
- Around line 175-177: Update upsert_datastore and BigQueryBackend.upsert so the
schema is read once within a backend-managed atomic operation and reused for
both _validate_records and the subsequent MERGE/INSERT DML. Remove the separate
pre-validation engine.info read, and pass the validated records or validation
callback into the atomic backend flow while preserving existing validation
behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 28b52a09-0765-4219-91b3-6a32c734820b
📒 Files selected for processing (2)
datastore/services/write.pytests/test_write_service.py
Comment on lines
+175
to
+177
| info = await asyncio.to_thread(engine.info, resource_id) | ||
| _validate_records(resource_id=resource_id, schema=info.schema, records=records) | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the service read/write boundary.
rg -n -C 8 'engine\.info|engine\.upsert|async def upsert_datastore' datastore/services/write.py
# Inspect engine implementations for transactions or shared synchronization
# that cover schema reads, schema changes, and upserts.
fd -t f -e py . datastore/infrastructure/engines | while IFS= read -r file; do
rg -n -C 5 'def (info|upsert|create|delete)\b|transaction|Lock|lock|mutex|semaphore' "$file" || true
doneRepository: datopian/datastore
Length of output: 7561
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the full service boundary and engine implementation around the relevant methods.
sed -n '140,195p' datastore/services/write.py
printf '\n--- BigQuery engine relevant methods ---\n'
sed -n '428,490p' datastore/infrastructure/engines/bigquery.py
sed -n '865,925p' datastore/infrastructure/engines/bigquery.py
sed -n '280,340p' datastore/infrastructure/engines/bigquery.py
printf '\n--- Shared lock/mutex candidates ---\n'
rg -n "\bclass .*Lock\b|\b[Aa]syncIO[Aa]ll\w*Lock\b|\bLock\(|\bMutex|threading\..*Lock|asyncio\.(Lock|Semaphore)|Semaphore|condition|Event|contextlib\." datastore/infrastructure datastore/services datastore/core datastore/api datastore/main.py || true
printf '\n--- Other upsert calls to get schema ---\n'
rg -n -C 5 'engine\.info\(|_validate_records\(|\.upsert\(' datastore/services/write.py datastore/api datastore/main.py datastore/tests 2>/dev/null || trueRepository: datopian/datastore
Length of output: 2181
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Files:\n'
git ls-files 'datastore/infrastructure/engines/*.py' 'datastore/services/write.py'
printf '\n--- BigQuery-related files ---\n'
fd -t f -e py 'bigquery|engine|engines' datastore/infrastructure datastore/core datastore/api datastore/services
printf '\n--- Upstream engine info/upsert definitions ---\n'
rg -n -C 8 'def (info|upsert|delete|create)\b|transaction|Lock|lock|mutex|semaphore|BEGIN|COMMIT|rollback' datastore/infrastructure/**/*.py datastore/services/*.py || true
printf '\n--- write.py relevant area ---\n'
sed -n '160,200p' datastore/services/write.py
printf '\n--- upsert calls and info usages ---\n'
rg -n -C 5 'engine\.info\(|\.upsert\(|_validate_records\(' datastore/services datastore/api datastore/main.py datastore/tests 2>/dev/null || trueRepository: datopian/datastore
Length of output: 12810
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- backend.py methods (info/upsert/metadata/schema alters) ---\n'
rg -n -C 10 'def (info|upsert|create|delete|_.*schema|_alter|_apply_schema|_validate|BEGIN|COMMIT|transaction|DELETE_COLUMN|SCHEMA|ALTER TABLE)' datastore/infrastructure/engines/bigquery/backend.py
printf '\n--- backend.py line count ---\n'
wc -l datastore/infrastructure/engines/bigquery/backend.py
printf '\n--- relevant backend slices ---\n'
sed -n '1,160p' datastore/infrastructure/engines/bigquery/backend.py
sed -n '340,470p' datastore/infrastructure/engines/bigquery/backend.py
sed -n '720,870p' datastore/infrastructure/engines/bigquery/backend.py
sed -n '870,950p' datastore/infrastructure/engines/bigquery/backend.pyRepository: datopian/datastore
Length of output: 247
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- backend.py methods (info/upsert/metadata/schema alters) ---'
rg -n -C 10 'def (info|upsert|create|delete|_.*schema|_alter|_apply_schema|_validate|BEGIN|COMMIT|transaction|DELETE_COLUMN|SCHEMA|ALTER TABLE)' datastore/infrastructure/engines/bigquery/backend.py || true
printf '%s\n' '--- backend.py line count ---'
wc -l datastore/infrastructure/engines/bigquery/backend.py
printf '%s\n' '--- relevant backend slices ---'
sed -n '1,160p' datastore/infrastructure/engines/bigquery/backend.py
sed -n '340,470p' datastore/infrastructure/engines/bigquery/backend.py
sed -n '720,950p' datastore/infrastructure/engines/bigquery/backend.py
printf '%s\n' '--- global lock/mutex/transaction references ---'
rg -n -C 5 'Lock\(|asyncio\.Lock|Semaphore|contextlib\.|threading\.[A-Za-z]*Lock|transaction|BEGIN|COMMIT|ROLLBACK|rollback|_lock|lock' datastore/infrastructure/engines datastore/services datastore/main.py || trueRepository: datopian/datastore
Length of output: 35854
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- BigQueryBackend.upsert full implementation ---'
sed -n '460,570p' datastore/infrastructure/engines/bigquery/backend.py
printf '%s\n' '--- BigQueryBackend.create relevant implementation ---'
sed -n '428,460p' datastore/infrastructure/engines/bigquery/backend.py
printf '%s\n' '--- merge DML/schema helpers ---'
sed -n '280,330p' datastore/infrastructure/engines/bigquery/lib.py
sed -n '950,1030p' datastore/infrastructure/engines/bigquery/lib.pyRepository: datopian/datastore
Length of output: 7356
Make the upsert schema read and the BigQuery write atomic.
datastore.services.write.upsert_datastore validates records with engine.info(resource_id).schema, but BigQueryBackend.upsert() reads the schema again before running MERGE/INSERT. A concurrent schema change can make the service validator apply to an old schema while the write succeeds with a different schema, or reject a row that would be valid under the current backend schema. Keep the schema read and the BigQuery DML in one backend-managed atomic block.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@datastore/services/write.py` around lines 175 - 177, Update upsert_datastore
and BigQueryBackend.upsert so the schema is read once within a backend-managed
atomic operation and reused for both _validate_records and the subsequent
MERGE/INSERT DML. Remove the separate pre-validation engine.info read, and pass
the validated records or validation callback into the atomic backend flow while
preserving existing validation behavior.
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.
Summary
datastore_createseed records anddatastore_upsertrecords against the stored Frictionless schema.Verification
uv run pytestuv run ruff check datastore testsSummary by CodeRabbit