refactor: convert SQLAlchemy sync to async (AsyncSession) - #151
Closed
wpfleger96 wants to merge 6 commits into
Closed
refactor: convert SQLAlchemy sync to async (AsyncSession)#151wpfleger96 wants to merge 6 commits into
wpfleger96 wants to merge 6 commits into
Conversation
… UTC contract Sync-prep PR for the two-PR async migration (see docs/async-migration-plan.md). The app stays fully synchronous; all changes are async-readiness foundations. Key changes: - Migrate all session.query() to select() + execute() (56 call sites, 14 files) - Add lazy="raise" to all ORM relationships; explicit eager loads at call sites - DatabaseTarget URL resolver with precedence chain and snore serve --db export - UTCDateTime custom type; absolute-instant columns migrated; wall-clock columns documented as naive by contract - SQLite connection recipe: connect_args autocommit=False + listener toggles dbapi_conn.autocommit=True around PRAGMAs, restores False (probe-verified) - Delete schema.sql (zero callers, obsolete SQLite-only subset) - DatabaseService SQL portability; Alembic env.py reads resolved URL - Transaction-ownership table; import savepoints with forced-failure test - I/O-compute DTO split; batch analysis splits into read+compute / write sessions to eliminate SQLite write-lock contention under ThreadPoolExecutor - Import-job state machine: pending->running->succeeded/failed/cancelled, pending->cancelled, coalescing observer channels, TTL reaping terminal-only - Docs: async-migration-plan.md, mcp-server-plan.md, roadmap.md entry Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Fixes all findings from Thufir's pass-1 review that were not yet addressed
in the working tree, plus mypy/ruff clean-up so just check passes green.
Mypy fixes:
- import_data.py: _sse_generator asserts isinstance(msg, dict) after
sentinel/None guards so mypy sees a concrete dict, not object
- test_sqlite_connection_recipe.py: _make_session_data return type changed
from object to UnifiedSession; __import__('datetime') replaced with
a direct import; TYPE_CHECKING guard added for the forward reference
- test_utc_datetime.py: dialect() constructor calls are untyped on some
SQLAlchemy builds; consolidated into named variables so the single
type: ignore comment covers the right line
Ruff fixes:
- export_service.py: orphaned 'from sqlalchemy import func' inside
_query_sessions removed (func was used in the old text() version,
not the typed select() version)
- analysis_facade.py: import block inside submit() re-sorted
- test_utc_datetime.py: import block inside AnalysisResult ordering test
re-sorted; removed duplicate dialect compile test methods
Drift test strengthened (§5):
- test_api_server.py: test_openapi_matches_committed_generated_types now
checks BOTH directions — generated types that no longer exist in the
live API (stale), AND live API paths missing from generated types
(ungenerated). Catches the exact failure Thufir named: a new DELETE
route whose types were never regenerated.
Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
…ricity, UTC, docs) Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
…ts, honest docs Pass-3 remediation batch addressing all 5 IMPORTANT + 2 MINOR findings from Thufir's final review of SNORE PR-1 (async-prep). ## §7 DTO/coordinator — bounded export streaming - ExportService._query_sessions_chunked: yields session dicts via yield_per() — no full fetchall() materialisation - ExportService._build_export_rows: generator that yields (session, events, settings) in EXPORT_CHUNK_SIZE=500 chunks; bulk-loads events+settings per chunk so memory is bounded regardless of result set size - export_csv and export_json consume _build_export_rows instead of _build_export_sessions; _build_export_sessions kept as backward-compat wrapper - analysis_facade.py: pending dict typed as dict[Future[str], int] (fixes mypy) - import_jobs.py state diagram corrected: POST starts the worker, not GET ## §6 Import ownership — deeper forced-failure test - test_sqlite_connection_recipe.py: new test_failed_session_after_child_writes injects failure AFTER waveform+event rows are flushed (at _import_statistics), asserts no device/day/session/waveform/event/statistics rows survive the savepoint rollback while the next session's rows do - _make_session_data_with_children: builds UnifiedSession with real waveform blob, event, statistics, and settings for rollback verification ## §3 Import lifecycle — shutdown test with live worker - test_import_state_machine.py: TestShutdown.test_shutdown_with_live_worker starts a real background thread blocked on a gate, calls shutdown(timeout=0.05), asserts _cancel_flag is set and 'still alive' warning is logged, then unblocks the thread and confirms it reaches terminal state ## §3/§4 Route/worker tests replacing simulations - TestRegistrationFailure: replaced manual simulation with three tests that call _start_worker directly; proves Thread.start() failure, Thread.__init__() failure, and create_job failure each cancel the job, remove it from the store, and clean up the upload directory - TestShutdown: added test_shutdown_with_live_worker (real thread, not pending-only) - TestRouteWorkerBehavior: new class with four route/worker tests: run_import_completes_as_succeeded, run_import_cancel_before_detect_produces_cancelled, sse_route_stream_emits_keepalive_while_running, route_cancel_delete_transitions_running - TestWorkerTerminalState docstring clarified: internal _finish/_finish_cancelled calls test the state machine; route-level behavior covered in new class ## §5 Honest behavior tests — UTC production path - test_utc_datetime.py: test_latest_result_via_production_path_with_mixed_offsets now calls AnalysisService.get_analysis_result() (the real service method) instead of reconstructing the query directly; distinguishes the two rows by session_duration_hours in programmatic_result_json ## §5 Honest behavior tests — serve --db lifespan test - test_database_target.py: TestServeCommandEnvExport.test_lifespan_opens_flagged_database runs the real serve Click command to capture the exported SNORE_DATABASE_URL, then invokes the FastAPI lifespan with that URL and asserts init_database_from_url is called with the flag-specified path — not just that the env var is set ## §5 Docs fixes - docs/async-migration-plan.md §6 ownership table: import_sessions_batch row corrected (caller provides session via db=, not opens own session_scope) - docs/async-migration-plan.md §6 footer: revised to describe actual call graph (ImportService injects session, import_sessions_batch uses savepoints only) - docs/async-migration-plan.md §7 batch analysis description: corrected 'read+compute session_scope' to three explicit phases (read-only, compute no-session, write-only) - docs/async-migration-plan.md §7 splits list: five → six surfaces; added Export (_build_export_rows generator); single-analysis description updated to load_session_inputs_raw + prepare_inputs + compute_analysis - docs/async-migration-plan.md §5: added dialect compilation suite mention and runtime-unproven disclaimer block - import_jobs.py module docstring state diagram: GET no longer starts the worker ## Test infrastructure fixes - test_analysis_batch.py: updated all monkeypatch targets from load_session_inputs/ _store_result to load_session_inputs_raw/prepare_inputs/store_result; cancel test now asserts cancelled==3, successful==0 (not trusts trivial load_calls==[]) - test_analysis_facade.py TestRunBatchAnalysis: same patch target updates - importers.py: batch_day_ids annotated as set[int] (mypy fix) - import_data.py: upload cleanup comment corrected to 'end of worker execution' (MINOR fix per spec) Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
PR-2 async flip dependencies. No greenlet pin. No Postgres drivers. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Replace sync Session with AsyncSession throughout services, API routers, CLI groups, and CLI commands. Converts all DB-touching code paths to use await session.execute() / await session.scalar() patterns. Key changes: - database/session.py: add async engine, session factory, session_scope(), cleanup_database(), open_db_session(); keep sync helpers for Alembic only - api/deps.py: replace sync get_db with async get_db yielding AsyncSession - All API routers: add async/await throughout - All service classes (session, device, day, event, stats, report, waveform, analysis_facade, database_service): converted to async methods - CLI groups: wrap async inner functions with asyncio.run() at click boundary - analysis_facade: get_analysis_result() queries DB directly (bypasses AnalysisService which is still sync-volatile) - api/routers/rx.py: rewritten to async using get_db - analysis/rx_tracker.py: all methods converted to async Test changes: - conftest.py: add async_db_session, async_test_device, async_test_session_factory - integration/conftest.py: autouse reset_database_state calls cleanup_database() - test_rx_tracker.py: all 25 tests converted to async def - test_analysis_facade.py: all 14 tests converted to async def - test_analysis_batch.py: mock execute as AsyncMock; coordinator test async - test_cli_commands.py: CLI tests use init_database + asyncio.run(setup) pattern instead of db_session (incompatible Alembic initialization paths); remove spurious init_database() from test_db_init_creates_database Volatile surfaces deferred per Paul's coordination note: - export_service.py (streaming rewrite in flight) - database_service.py is_sqlite_target()/reset() (split pending) - import_jobs.py shutdown/reaper (stop-event semantics pending) - import_service.py/importers.py transaction ownership (per-chunk scopes pending) - analysis_facade.py batch ID discovery / single/waveform/report read-scope - _make_compute_only() (public seam replacement pending) Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
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.
What
Converts the entire SQLAlchemy usage from synchronous
Sessionto asyncAsyncSession(aiosqlite backend).Why
The API layer (FastAPI), CLI bridge (
asyncio.run()wrappers), and analysis pipeline all run in async contexts. Using a sync session forcedrun_syncescape hatches and blocked the event loop during DB I/O. This conversion makes the session type consistent with how every caller already expects to use it.Scope
Production changes:
database/session.py: async engine,AsyncSessionFactory,session_scope(),cleanup_database(),open_db_session(); sync helpers kept for Alembic onlyapi/deps.py:get_dbnow yieldsAsyncSessionasync def+await session.execute()throughoutsession_service,device_service,day_service,event_service,stats_service,report_service,waveform_service,analysis_facade,database_service): methods converted to async_run()functions wrapped withasyncio.run()at the Click boundaryanalysis_facade.get_analysis_result(): queries DB directly — bypassesAnalysisServicewhich is still volatileapi/routers/rx.py: rewritten async usingget_dbanalysis/rx_tracker.py: all methods asyncTest changes:
conftest.py:async_db_session,async_test_device,async_test_session_factoryfixturesintegration/conftest.py:autousereset_database_statecallscleanup_database()before/after each testtest_rx_tracker.py: all 25 tests →async deftest_analysis_facade.py: all 14 tests →async deftest_analysis_batch.py:mock.execute→AsyncMock; coordinator test asynctest_cli_commands.py: CLI tests useinit_database+asyncio.run(setup)instead ofdb_sessionfixture (fixes Alembic/create_alldouble-initialization conflict)Volatile surfaces deferred
Per coordination note — Duncan's
duncan/async-prepbranch rewrites these; diffs kept minimal:export_service.py— streaming rewrite in flightdatabase_service.pyis_sqlite_target()/reset()— capability split pendingimport_jobs.pyshutdown/reaper — stop-event semantics pendingimport_service.py/importers.py— per-chunk injected scope ownership pendinganalysis_facade.pybatch ID discovery, single/waveform/report read-scope — streaming pending_make_compute_only()— public seam replacement pendingResult
1063 passed, 61 skipped, 0 failures.