One useful idea
A dictionary survives only as long as its process. A repository reads and writes an actual database through a small interface; restarting it must recover stored values, not return a canned report. This kit uses SQLite directly so you can see parameterized statements and transactions before considering an optional ORM. The original source build identity is preserved; SQLAlchemy is not installed or silently required.
Review the path/connection policy and migration statements, then own their orchestration in Repository. Use an explicit owned .db path under an existing parent. initialize(new=True) creates exclusively; existing NEW destinations, missing parents, traversal, reserved/ambiguous paths, known links/reparse points and oversized existing files refuse. Ordinary initialize() opens an existing bounded file, checks application marker 1146507280/user_version and refuses foreign/future databases without migrating them. This is not hostile-filesystem race isolation.
Version 1 has evaluations/audit_events; version 2 adds findings. Execute pending declared migration statements individually under BEGIN IMMEDIATE, updating user_version only with the transaction. Do not use executescript here: its transaction behavior would undermine this explicitly controlled migration. Repeating supported initialization must be idempotent. A failed NEW initialization may leave a visible partial NEW file; inspect it and choose another new name. Never automatically delete, replace or reset a learner database.
Bind values with SQL parameters, never f-string evidence into a query. put_evaluation and add_finding write the record AND controlled event before COMMIT; any fault causes ROLLBACK and re-raise. get methods parse actual stored JSON back into reviewed models. Scope finding lookup to both parent and finding ID. apply_transition reads the actual stored record, invokes the declared transition callback, verifies unchanged identity/evidence/rating and performs UPDATE with an expected revision condition; require rowcount 1, then audit in that same transaction. A stale update or failed event must not leave a changed finding or extra event.
The reviewed connection helper closes in finally. A sqlite3 connection’s own with-block controls commit/rollback, not connection closure; use explicit closing when inspecting with the raw library. Fresh audit dictionaries prevent a caller from mutating later reads. Complete fixture histories grow in memory, and the existing-file budget is 16 MiB: there is no production pagination/load guarantee. Keep small invented fixtures, not real customer material.
The example performs actual v1→v2 migration, persisted typed reopen and an injected audit-write failure. TemporaryDirectory owns only its demonstration files. Those observations do not prove a public service, authenticated authorship, general backup strategy or completed independent portfolio.
Refresh first: Typed stored evaluations, Failures must remain visible, SQLite tables and parameterized queries, Actual file integration versus mocks.
Trace a finished example
from pathlib import Path
from tempfile import TemporaryDirectory
import sqlite3
from evaluation_service.app import FIXTURE_CREDENTIALS
from evaluation_service.core import build_evaluation
from evaluation_service.database import create_v1, connection
from evaluation_service.errors import Missing
from evaluation_service.models import EvaluationCreate
from evaluation_service.storage import Repository
with TemporaryDirectory(prefix="dvp-owned-migration-") as folder:
path = Path(folder) / "example.db"
create_v1(path) # Reviewed owned historical fixture, not a learner migration.
with connection(path) as database:
before = database.execute("PRAGMA user_version").fetchone()[0]
store = Repository(path)
store.initialize()
with connection(path) as database:
after = database.execute("PRAGMA user_version").fetchone()[0]
print(before, "->", after)
request = EvaluationCreate(asset_id="Fresh", width=1920, height=1080, frames=240, fps=24)
actor = FIXTURE_CREDENTIALS["fixture-author-a"]
value = build_evaluation(request, actor, "evaluation_A")
store.put_evaluation(value)
reopened = Repository(path)
reopened.initialize()
print(reopened.get_evaluation(value.id) == value, len(reopened.audit(value.id)))
with connection(path) as database:
database.execute("CREATE TRIGGER stop_event BEFORE INSERT ON audit_events "
"BEGIN SELECT RAISE(ABORT,'invented fault'); END")
try:
reopened.put_evaluation(build_evaluation(request, actor, "evaluation_B"))
except sqlite3.IntegrityError:
try:
reopened.get_evaluation("evaluation_B")
except Missing:
print("Missing", len(reopened.audit(value.id)))Expected output
1 -> 2
True 1
Missing 1The declared owned v1 file is really migrated to v2. A fresh repository reads the same typed evaluation and its single committed event. A trigger aborts the next event insertion, so that evaluation insert rolls back too; it is missing and the earlier evaluation still has one event. No fake write or success receipt substitutes for these reads.
The finished implementation is in evaluation_service/storage.py. Reading it is guided practice, not independent evidence.
Predict an atomic failure
The evaluation INSERT works but its audit INSERT fails. Which record/event should remain?
Compare your answer · self-reviewed
Neither new write should remain. BEGIN/COMMIT cover both; rollback and re-raise the event fault. Earlier committed records/events remain unchanged.
Find unsafe SQL
Why is INSERT ... VALUES(?) with a parameter better than formatting evidence into a query?
Compare your answer · self-reviewed
The parameter remains a value, including quotes or SQL-looking text. String interpolation can change SQL structure. This does not make unrelated filesystem/authentication policy automatically safe.
Recall resource ownership
Does with sqlite3.connect(path) automatically close the connection and certify persistence?
Compare your answer · self-reviewed
No. That context controls a transaction, not closure. Close the connection explicitly and reopen/read actual typed values to observe persistence; a printed promise is not evidence.
Change it, then build your own
One controlled change
Use create_v1 only for a NEW owned historical fixture, then initialize twice and inspect actual version 2. Insert changed supplied values and reopen. In a separate owned database inject an audit trigger failure; confirm no partial record/event survives.
Your independent task
Implement every Repository method in practice.py using the reviewed checked_path/connection/SCHEMA assistance, but not the finished Repository. Own migration iteration, actual parameterized writes/reads, exact NEW flag, explicit commit/rollback, fresh audit output and atomic compare-and-update revision. Keep record/evidence identity and parent scope. Run selected Build 3 tests/demo; this group includes real event-failure/migration/restart observations.
What success looks like
Supported v1→v2 migration is real and idempotent; foreign/future files refuse unchanged. Typed values persist after reopening. Duplicate/missing/unsafe inputs visibly refuse; injected evaluation/finding/transition event failures roll back associated changes. SQL-looking evidence stays data. A second init at an existing NEW path never overwrites it.
Hint 1 · a question
Mark each operation’s read, record write, audit write and commit. Which pair must succeed or fail together?
Hint 2 · a concept cue
Use BEGIN IMMEDIATE, execute parameterized statements, commit only after all required writes; rollback then re-raise on failure. Close every connection through the reviewed helper.
Hint 3 · a localized example
A scoped update includes id, evaluation_id and expected revision in its WHERE clause. Require exactly one changed row before committing the matching controlled event.
Need the complete worked solution?
Open evaluation_service/storage.py from the kit. Trace it, close it, then try fresh inputs in your own files. Treat the attempt as guided; seeing the solution does not award a practical pass.
Course help is guidance, not independent evidence. With JavaScript, opening help records guidance locally; otherwise note it in your README. Reset does not erase that history.
Repair a failed check
If a reopen loses values, inspect actual INSERT/COMMIT and the database path rather than printing a receipt. If an audit fault leaves a record, expand the transaction to both writes. If stale updates win, condition the actual UPDATE on expected revision and check rowcount. If Windows cleanup reports a locked inspection file, close the raw connection explicitly—do not suppress cleanup errors or delete another database.
NotImplementedError means a practice stub is still unfinished. Read the failing test name and the last error line. Change one behavior, rerun that build, then rerun all implemented builds.
Show it works on new inputs
Create a new owned v1 fixture, migrate/reinitialize, persist changed evaluations/findings, reopen and retain actual SQL/version/typed read observations. Force one audit failure and one stale update; show unchanged prior state and no extra event. Explain helper assistance, possible partial NEW initialization and why this is not a production migration/backup system.
Self-review: name the input, result, refused case and reason. Your local test output and explanation are separate from a quiz score; this page does not certify a pass.
Keep the idea
Persistence is an observed read after a write and restart. A transaction makes related writes one outcome; parameterized values and revision guards protect the declared operation.