diff --git a/pyproject.toml b/pyproject.toml index dbd11dfd..4805d1c2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,7 @@ classifiers = [ dependencies = [ "fastapi~=0.135", "uvicorn[standard]~=0.34", - "sqlalchemy[asyncio]~=2.0.0", # Keep the qualified 2.0 typing/runtime contract. + "sqlalchemy[asyncio]>=2.1.4,<2.2", # Keep the qualified 2.1 typing/runtime contract. "alembic~=1.14", "mako>=1.3.12", # transitive via alembic — floor for CVE-2026-44307 "aiosqlite~=0.20", diff --git a/src/pullbox/core/cover_migration.py b/src/pullbox/core/cover_migration.py index 75c7a678..82d3b28a 100644 --- a/src/pullbox/core/cover_migration.py +++ b/src/pullbox/core/cover_migration.py @@ -9,7 +9,7 @@ import shutil from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast import structlog @@ -38,10 +38,10 @@ async def migrate_covers_to_dotcovers( async with session_factory() as session: # Check if already migrated - result = await session.execute( + migration_result = await session.execute( select(SystemConfig.value).where(SystemConfig.key == _MIGRATION_KEY) ) - if result.scalar_one_or_none() == "true": + if migration_result.scalar_one_or_none() == "true": return 0 # Resolve target covers directory @@ -62,7 +62,8 @@ async def migrate_covers_to_dotcovers( moved = 0 for series_id, series_path_str in series_rows: - series_path = Path(series_path_str) + # The query excludes NULL paths; SQL predicates do not narrow row types. + series_path = Path(cast("str", series_path_str)) if not series_path.is_dir(): continue diff --git a/src/pullbox/core/scheduler_stats_persistence.py b/src/pullbox/core/scheduler_stats_persistence.py index 29676008..9f577b55 100644 --- a/src/pullbox/core/scheduler_stats_persistence.py +++ b/src/pullbox/core/scheduler_stats_persistence.py @@ -9,9 +9,9 @@ from typing import Any import structlog -from sqlalchemy import delete, func, select +from sqlalchemy import delete, func, make_url, select from sqlalchemy.dialects.sqlite import insert as sqlite_insert -from sqlalchemy.exc import DatabaseError, OperationalError +from sqlalchemy.exc import ArgumentError, DatabaseError, OperationalError from pullbox.core.scheduler_error_helpers import ( is_locked_error, @@ -190,15 +190,21 @@ def resolve_runtime_db_url(default_db_url: str, session_factory: Any) -> str: def resolve_legacy_task_stats_sidecar_path(db_url: str) -> Path | None: """Return the retired JSON sidecar path for one-time import.""" - if "sqlite" not in db_url or ":memory:" in db_url: + try: + url = make_url(db_url) + except ArgumentError: + return None + if url.get_backend_name() != "sqlite" or not url.database or url.database == ":memory:": return None - if ":///" in db_url: - raw_path = db_url.split(":///", 1)[1] - elif "://" in db_url: - raw_path = db_url.split("://", 1)[1] - else: + uri = url.query.get("uri") + if ( + url.database.startswith("file:") + and isinstance(uri, str) + and uri.strip().lower() in {"true", "1", "yes", "on", "y", "t"} + and (url.database == "file::memory:" or url.query.get("mode") == "memory") + ): return None - db_path = Path(raw_path.split("?", 1)[0]) + db_path = Path(url.database) return db_path.with_name("scheduled_task_stats.json") diff --git a/src/pullbox/services/direct_discovery_retention.py b/src/pullbox/services/direct_discovery_retention.py index 7103906e..20e52cda 100644 --- a/src/pullbox/services/direct_discovery_retention.py +++ b/src/pullbox/services/direct_discovery_retention.py @@ -20,7 +20,7 @@ async def prune_unstarted_direct_discoveries( session: AsyncSession, - search_log_ids: Select[tuple[int]] | Sequence[int], + search_log_ids: Select[int] | Sequence[int], ) -> int: """Delete only discoveries that never progressed beyond search results.""" result = await session.execute( diff --git a/src/pullbox/services/import_deferred_recovery.py b/src/pullbox/services/import_deferred_recovery.py index de086bfb..cc39bdbc 100644 --- a/src/pullbox/services/import_deferred_recovery.py +++ b/src/pullbox/services/import_deferred_recovery.py @@ -329,8 +329,8 @@ async def plan_deferred_recovery( continue elif provider_id is None and item.series_id in series_by_id: series = series_by_id[item.series_id] - issue = strict_filename_target(file, item, series, issues_by_series[series.id]) - target = (issue, series) if issue is not None else None + filename_issue = strict_filename_target(file, item, series, issues_by_series[series.id]) + target = (filename_issue, series) if filename_issue is not None else None registered_file = registered.get(path) if registered_file is not None: @@ -633,7 +633,7 @@ async def refresh_recovered_groups( await recompute_series_counters(session, job) -def empty_stale_series_query(job_id: int) -> Select[tuple[ImportedSeries]]: +def empty_stale_series_query(job_id: int) -> Select[ImportedSeries]: """Return empty missing-location rows eligible for follow-up archival.""" return select(ImportedSeries).where( ImportedSeries.import_job_id == job_id, diff --git a/src/pullbox/services/import_file_match_targets.py b/src/pullbox/services/import_file_match_targets.py index 53e89fb5..9342c709 100644 --- a/src/pullbox/services/import_file_match_targets.py +++ b/src/pullbox/services/import_file_match_targets.py @@ -97,7 +97,13 @@ async def load_file_match_target_index( for issue, library_file_id in issues_result.all(): has_library_file = library_file_id is not None target_index.issue_entries.append((issue, has_library_file)) - entry = (issue.id, issue.comicvine_id, has_library_file, issue, issue.title) + entry: FileMatchTargetEntry = ( + issue.id, + issue.comicvine_id, + has_library_file, + issue, + issue.title, + ) if issue.comicvine_id is not None: target_index.cv_id_map[issue.comicvine_id] = entry target_index.exact_number_map[issue.effective_issue_number_text] = entry diff --git a/src/pullbox/services/import_file_matching.py b/src/pullbox/services/import_file_matching.py index 53790eb6..ad9e49e5 100644 --- a/src/pullbox/services/import_file_matching.py +++ b/src/pullbox/services/import_file_matching.py @@ -78,6 +78,7 @@ scan_review_file_target_weight, scan_review_progress_pct, ) +from pullbox.services.import_provider_cache import is_in_memory_sqlite_engine from pullbox.services.import_provisional_targets import reconcile_provisional_targets from pullbox.services.import_source_metadata import ( corroborated_import_title_conflict, @@ -588,9 +589,8 @@ def write_rows(rows: Any) -> None: await session.commit() await raise_if_cancelled(session, job_id) bind = session.bind - bind_url = str(bind.sync_engine.url) if bind is not None else "" execution_options = {"pullbox_cohort_scan": scan_name} - if bind is None or ":memory:" in bind_url: + if bind is None or is_in_memory_sqlite_engine(bind): result = await session.execute(query.execution_options(**execution_options)) for rows in result.partitions(size): await raise_if_cancelled(session, job_id) diff --git a/src/pullbox/services/import_job_controls.py b/src/pullbox/services/import_job_controls.py index f113fc55..55ce4f08 100644 --- a/src/pullbox/services/import_job_controls.py +++ b/src/pullbox/services/import_job_controls.py @@ -16,6 +16,7 @@ sqlite_lock_retry_delay, ) from pullbox.models.import_job import ImportControlRequest, ImportJob, ImportJobStatus +from pullbox.services.import_provider_cache import is_in_memory_sqlite_engine from pullbox.services.import_workflow_state import ( initialize_progress_snapshot, paused_message_for_mode, @@ -100,8 +101,7 @@ async def raise_if_job_cancelled_immediately( await raise_if_cancelled(session, job_id) return - bind_url = str(bind.sync_engine.url) - if ":memory:" in bind_url: + if is_in_memory_sqlite_engine(bind): with session.no_autoflush: await raise_if_cancelled(session, job_id) return diff --git a/src/pullbox/services/import_job_execution_progress.py b/src/pullbox/services/import_job_execution_progress.py index 41a26ffb..6b5d13b5 100644 --- a/src/pullbox/services/import_job_execution_progress.py +++ b/src/pullbox/services/import_job_execution_progress.py @@ -36,6 +36,7 @@ import_group_progress_plan, import_work_progress, ) +from pullbox.services.import_provider_cache import is_in_memory_sqlite_engine if TYPE_CHECKING: from collections.abc import Awaitable, Callable @@ -151,8 +152,7 @@ def progress_session_factory_for_runtime( bind = session.bind if bind is None: return None - bind_url = str(bind.sync_engine.url) - if ":memory:" in bind_url: + if is_in_memory_sqlite_engine(bind): return None return async_sessionmaker( bind, diff --git a/src/pullbox/services/import_orphans.py b/src/pullbox/services/import_orphans.py index 1191ecd0..3aa48a37 100644 --- a/src/pullbox/services/import_orphans.py +++ b/src/pullbox/services/import_orphans.py @@ -565,7 +565,7 @@ async def recover_orphan( return recovery_summary -def _orphaned_series_query(*, job_id: int | None = None) -> Select[tuple[ImportedSeries]]: +def _orphaned_series_query(*, job_id: int | None = None) -> Select[ImportedSeries]: """Build the shared unresolved-orphan filter used by list and count queries.""" query = ( sa_select(ImportedSeries) diff --git a/src/pullbox/services/import_path_reconciliation.py b/src/pullbox/services/import_path_reconciliation.py index df5085e6..ba3b236f 100644 --- a/src/pullbox/services/import_path_reconciliation.py +++ b/src/pullbox/services/import_path_reconciliation.py @@ -4,7 +4,7 @@ import asyncio from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from sqlalchemy import func, or_, select from sqlalchemy.orm import aliased @@ -41,7 +41,7 @@ def _candidate_query( job_id: int, -) -> tuple[Select[tuple[ImportedFile, ImportedFile, ImportedSeries]], type[ImportedFile]]: +) -> tuple[Select[ImportedFile, ImportedFile, ImportedSeries], type[ImportedFile]]: # Count all same-folder copies, including blocked ones, before choosing a match. folder = func.substr( ImportedFile.file_path, @@ -241,7 +241,8 @@ async def reconcile_saved_mylar_paths( **reconciliation_evidence( record.file_path, actual.file_path, - record.comicvine_issue_id, + # Both candidate CTEs exclude NULL provider IDs. + cast("int", record.comicvine_issue_id), recorded_series_name=base.series_name, actual_series_name=metadata.series_name, ), @@ -271,13 +272,13 @@ async def reconcile_saved_mylar_paths( if apply and report["references_reconciled"]: await recompute_file_counters(session, job) for series_id in changed_series: - series = await session.get(ImportedSeries, series_id) - if series is not None: - series.file_count = series.files_total - series.sample_paths = list( + changed_series_row = await session.get(ImportedSeries, series_id) + if changed_series_row is not None: + changed_series_row.file_count = changed_series_row.files_total + changed_series_row.sample_paths = list( await session.scalars( select(ImportedFile.file_path) - .where(ImportedFile.import_series_id == series.id) + .where(ImportedFile.import_series_id == changed_series_row.id) .order_by(ImportedFile.id) .limit(5) ) diff --git a/src/pullbox/services/import_pending_recovery.py b/src/pullbox/services/import_pending_recovery.py index 8f2c8def..f9c9596b 100644 --- a/src/pullbox/services/import_pending_recovery.py +++ b/src/pullbox/services/import_pending_recovery.py @@ -22,7 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession -def pending_file_ids(job_id: int) -> Select[tuple[int]]: +def pending_file_ids(job_id: int) -> Select[int]: """Only already-confirmed targets, never unresolved or unselected siblings.""" return ( select(ImportedFile.id) diff --git a/src/pullbox/services/import_provider_cache.py b/src/pullbox/services/import_provider_cache.py index 29d4bf23..a802aa2b 100644 --- a/src/pullbox/services/import_provider_cache.py +++ b/src/pullbox/services/import_provider_cache.py @@ -24,20 +24,33 @@ if TYPE_CHECKING: from collections.abc import Callable, Coroutine, Sequence - from sqlalchemy.ext.asyncio import AsyncSession + from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession _T = TypeVar("_T") -def is_in_memory_sqlite_engine(engine: AsyncEngine) -> bool: +def is_in_memory_sqlite_engine(engine: AsyncEngine | AsyncConnection) -> bool: """Return True when separate sessions would share one SQLite memory connection.""" - url = engine.url + url = engine.sync_engine.url if url.get_backend_name() != "sqlite": return False database = url.database if database in {None, "", ":memory:"}: return True - return str(database).startswith("file:") and "mode=memory" in str(url.query) + uri = url.query.get("uri", "false") + uri_enabled = isinstance(uri, str) and uri.strip().lower() in { + "true", + "yes", + "on", + "y", + "t", + "1", + } + return ( + uri_enabled + and database.startswith("file:") + and (database == "file::memory:" or url.query.get("mode") == "memory") + ) def build_import_scan_metadata_provider( diff --git a/src/pullbox/services/import_reference_recovery.py b/src/pullbox/services/import_reference_recovery.py index cad50cff..0c07e6c9 100644 --- a/src/pullbox/services/import_reference_recovery.py +++ b/src/pullbox/services/import_reference_recovery.py @@ -52,7 +52,7 @@ from pullbox.services.metadata_service import MetadataService -def reference_candidate_ids(job_id: int) -> Select[tuple[int]]: +def reference_candidate_ids(job_id: int) -> Select[int]: """Bound the existing recheck preview to potentially misplaced references.""" return ( select(ImportedFile.id) diff --git a/src/pullbox/services/import_series_deduplication.py b/src/pullbox/services/import_series_deduplication.py index 0f0daf92..9fb1d9f9 100644 --- a/src/pullbox/services/import_series_deduplication.py +++ b/src/pullbox/services/import_series_deduplication.py @@ -429,16 +429,14 @@ async def _load_existing_name_candidate_page( def _existing_series_candidate_select() -> Select[ - tuple[ - int, - int | None, - str, - int | None, - SeriesType, - str, - int, - str | None, - ] + int, + int | None, + str, + int | None, + SeriesType, + str | None, + int, + str | None, ]: return sa_select( Series.id, diff --git a/src/pullbox/services/metadata_identity_inventory.py b/src/pullbox/services/metadata_identity_inventory.py index d9980291..9e7d757f 100644 --- a/src/pullbox/services/metadata_identity_inventory.py +++ b/src/pullbox/services/metadata_identity_inventory.py @@ -155,12 +155,14 @@ async def read_legacy_identity_page( "Story arc identity row budget exceeded. Smaller pages may help; " "an arc that exceeds the budget alone needs a separate inspection." ) - for row_id, local_id, source, value in arc_rows: + for row_id, local_id, source, external_id in arc_rows: storage = LegacyIdentityStorage.STORY_ARC_RELATION try: if source == IdentityNamespace.COMICVINE: - value = normalize_comicvine_arc_id(value) - identity = ExternalIdentityRef(IdentityNamespace(source), entity_kind, value) + external_id = normalize_comicvine_arc_id(external_id) + identity = ExternalIdentityRef( + IdentityNamespace(source), entity_kind, external_id + ) except ValueError: problems.append( LegacyIdentityProblem(local_id, storage, row_id, "invalid_external_id") diff --git a/src/pullbox/services/reading_query_service.py b/src/pullbox/services/reading_query_service.py index 1b4f01f0..0fbe0489 100644 --- a/src/pullbox/services/reading_query_service.py +++ b/src/pullbox/services/reading_query_service.py @@ -576,7 +576,7 @@ def _adjacent_statement( series_id: int, boundary: ColumnElement[bool], descending: bool, -) -> Select[tuple[int, float, str | None]]: +) -> Select[int, float, str | None]: issue_order = Issue.issue_number.desc() if descending else Issue.issue_number.asc() id_order = Issue.id.desc() if descending else Issue.id.asc() return ( @@ -599,7 +599,7 @@ def _adjacent_statement( def _adjacent_reference( - row: Row[tuple[int, float, str | None]] | None, + row: Row[int, float, str | None] | None, ) -> AdjacentIssueReference | None: if row is None: return None diff --git a/src/pullbox/services/search_runtime.py b/src/pullbox/services/search_runtime.py index 3efa09a1..fa9fd503 100644 --- a/src/pullbox/services/search_runtime.py +++ b/src/pullbox/services/search_runtime.py @@ -139,7 +139,7 @@ async def build_search_runtime( ) ) ) - cfg: dict[str, str] = dict(cfg_rows.all()) # type: ignore[arg-type] + cfg: dict[str, str] = dict(cfg_rows.all()) try: source_priority_raw = json.loads(cfg.get("source_priority", "[]")) diff --git a/src/pullbox/services/story_arc_managed_reorder.py b/src/pullbox/services/story_arc_managed_reorder.py index f6ebcd81..8f0f650d 100644 --- a/src/pullbox/services/story_arc_managed_reorder.py +++ b/src/pullbox/services/story_arc_managed_reorder.py @@ -1113,7 +1113,7 @@ def _serializer() -> URLSafeTimedSerializer: def _pending_coordinator_statement( story_arc_id: int, -) -> Select[tuple[StoryArcPlacement]]: +) -> Select[StoryArcPlacement]: """Return the bounded, portable durable-reorder discovery query.""" return ( select(StoryArcPlacement) diff --git a/src/pullbox/services/story_arc_policy_migration.py b/src/pullbox/services/story_arc_policy_migration.py index dd6c0a77..2c61fa89 100644 --- a/src/pullbox/services/story_arc_policy_migration.py +++ b/src/pullbox/services/story_arc_policy_migration.py @@ -72,6 +72,21 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import Select + _MembershipScopeValues = tuple[IssueStoryArc, Issue | None, Series | None, Publisher | None] + _TerminalSyncWorkValues = tuple[ + int, + object, + int | None, + int | None, + int, + str, + int | None, + ImportJobStatus | None, + int | None, + int | None, + ImportJobActionStatus | None, + ] + STORY_ARC_POLICY_MIGRATION_CONFIRMATION = "CHANGE STORY ARC PLACEMENT POLICY" _TOKEN_SALT = "pullbox-story-arc-policy-migration-v1" @@ -787,7 +802,7 @@ def _membership_scope_statement( *, after_membership_id: int, limit: int, -) -> Select[tuple[IssueStoryArc, Issue, Series, Publisher]]: +) -> Select[*_MembershipScopeValues]: """Portable, bounded keyset query used by every membership digest pass.""" return ( select(IssueStoryArc, Issue, Series, Publisher) @@ -808,7 +823,7 @@ def _placement_scope_statement( *, after_placement_id: int, limit: int, -) -> Select[tuple[StoryArcPlacement]]: +) -> Select[StoryArcPlacement]: """Portable, bounded keyset query for the exact placement scope.""" return ( select(StoryArcPlacement) @@ -822,7 +837,7 @@ def _placement_scope_statement( ) -def _active_operation_statement(story_arc_id: int) -> Select[tuple[int]]: +def _active_operation_statement(story_arc_id: int) -> Select[int]: """Fail closed on both durable tokens and malformed recovery status rows.""" return ( select(StoryArcPlacement.id) @@ -839,7 +854,7 @@ def _active_operation_statement(story_arc_id: int) -> Select[tuple[int]]: ) -def _active_sync_work_statement(story_arc_id: int) -> Select[tuple[int]]: +def _active_sync_work_statement(story_arc_id: int) -> Select[int]: return ( select(StoryArcSyncWork.id) .join(IssueStoryArc, StoryArcSyncWork.issue_story_arc_id == IssueStoryArc.id) @@ -860,7 +875,7 @@ def _terminal_sync_work_statement( *, after_work_id: int, limit: int, -) -> Select[tuple[object, ...]]: +) -> Select[*_TerminalSyncWorkValues]: """Bound terminal work so Python can validate nested rollback markers exactly.""" return ( select( @@ -921,7 +936,7 @@ async def _active_sync_recovery_exists( await asyncio.sleep(0) -def _terminal_sync_scope(row: Row[tuple[object, ...]]) -> _TerminalSyncWorkScope: +def _terminal_sync_scope(row: Row[*_TerminalSyncWorkValues]) -> _TerminalSyncWorkScope: ( work_id, last_result, @@ -936,17 +951,17 @@ def _terminal_sync_scope(row: Row[tuple[object, ...]]) -> _TerminalSyncWorkScope action_status, ) = row return _TerminalSyncWorkScope( - id=cast("int", work_id), + id=work_id, last_result=last_result, - origin_import_job_id=cast("int | None", origin_import_job_id), - origin_import_action_id=cast("int | None", origin_import_action_id), - issue_story_arc_id=cast("int", membership_id), - desired_generation=cast("str", desired_generation), - job_id=cast("int | None", job_id), - job_status=cast("ImportJobStatus | None", job_status), - action_id=cast("int | None", action_id), - action_import_job_id=cast("int | None", action_import_job_id), - action_status=cast("ImportJobActionStatus | None", action_status), + origin_import_job_id=origin_import_job_id, + origin_import_action_id=origin_import_action_id, + issue_story_arc_id=membership_id, + desired_generation=desired_generation, + job_id=job_id, + job_status=job_status, + action_id=action_id, + action_import_job_id=action_import_job_id, + action_status=action_status, ) @@ -1225,7 +1240,7 @@ async def _membership_scopes_from_rows( session: AsyncSession, *, arc_name: str, - rows: Sequence[Row[tuple[IssueStoryArc, Issue, Series, Publisher]]], + rows: Sequence[Row[*_MembershipScopeValues]], ) -> tuple[_MembershipScope, ...]: issue_ids = tuple(issue.id for _membership, issue, _series, _publisher in rows if issue) files_by_issue: dict[int, LibraryFile] = {} diff --git a/src/pullbox/ui/import_follow_up.py b/src/pullbox/ui/import_follow_up.py index 3d0a050b..7732f125 100644 --- a/src/pullbox/ui/import_follow_up.py +++ b/src/pullbox/ui/import_follow_up.py @@ -123,7 +123,7 @@ def _follow_up_job_filter() -> ColumnElement[bool]: ) -def _follow_up_jobs_statement() -> Select[tuple[ImportJob]]: +def _follow_up_jobs_statement() -> Select[ImportJob]: return select(ImportJob).where( ImportJob.status.in_(_FOLLOW_UP_JOB_STATUSES), ImportJob.archived_at.is_(None), diff --git a/src/pullbox/ui/import_review_lanes.py b/src/pullbox/ui/import_review_lanes.py index 9a588ec7..d1b4111f 100644 --- a/src/pullbox/ui/import_review_lanes.py +++ b/src/pullbox/ui/import_review_lanes.py @@ -4,7 +4,7 @@ from collections import Counter, defaultdict from dataclasses import dataclass, field -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast from sqlalchemy import func, select @@ -286,8 +286,9 @@ async def _identity_conflict_series_ids(session: AsyncSession, job_id: int) -> s .distinct() ) for group_id, series_id, title, year, content_hash, file_name in records: + # The query excludes NULL group IDs; retain that fact at the typed boundary. series, titles, saved_titles, years, hashes = groups.setdefault( - group_id, (set(), set(), set(), set(), set()) + cast("int", group_id), (set(), set(), set(), set(), set()) ) series.add(series_id) if title: diff --git a/src/pullbox/ui/intervention_context_loaders.py b/src/pullbox/ui/intervention_context_loaders.py index d3befe5e..a31ced93 100644 --- a/src/pullbox/ui/intervention_context_loaders.py +++ b/src/pullbox/ui/intervention_context_loaders.py @@ -3,7 +3,6 @@ from __future__ import annotations from typing import TYPE_CHECKING -from typing import cast as typing_cast from sqlalchemy import ColumnElement, String, cast, func, or_, select from sqlalchemy.orm import joinedload @@ -144,8 +143,7 @@ async def load_intervention_queue_context( .group_by(PendingMatch.confidence) ) confidence_counts = { - (typing_cast("str", confidence_value) or "").lower(): count - for confidence_value, count in summary_result.all() + (confidence_value or "").lower(): count for confidence_value, count in summary_result.all() } pending_count = sum(confidence_counts.values()) @@ -278,8 +276,7 @@ async def load_intervention_history_context( .group_by(PendingMatch.status) ) status_counts = { - (typing_cast("str", status_value) or "").lower(): count - for status_value, count in summary_result.all() + (status_value or "").lower(): count for status_value, count in summary_result.all() } filtered_count: int = ( @@ -376,8 +373,7 @@ async def load_intervention_context( .group_by(PendingMatch.confidence) ) confidence_counts = { - (typing_cast("str", confidence_value) or "").lower(): count - for confidence_value, count in summary_result.all() + (confidence_value or "").lower(): count for confidence_value, count in summary_result.all() } active_pending_count = sum(confidence_counts.values()) match_review_count: int = ( diff --git a/tests/integration/metadata_identity/test_file_mutation_admission_autoflush.py b/tests/integration/metadata_identity/test_file_mutation_admission_autoflush.py new file mode 100644 index 00000000..8870c070 --- /dev/null +++ b/tests/integration/metadata_identity/test_file_mutation_admission_autoflush.py @@ -0,0 +1,101 @@ +"""Autoflush retains admission ordering for clean and already-locked callers.""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import TYPE_CHECKING, Any + +import pytest +from sqlalchemy import event, func, select + +from pullbox.models import Series +from pullbox.services import library_mutation_coordination as coordination + +if TYPE_CHECKING: + from collections.abc import Iterator + + from sqlalchemy import Select + from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession + + from tests.fixtures.metadata_identity_persistence import IdentityProbeDatabase + +pytestmark = pytest.mark.parametrize("identity_probe_db", ["postgresql"], indirect=True) + + +@contextmanager +def _record_admission_and_writes(engine: AsyncEngine) -> Iterator[list[str]]: + operations: list[str] = [] + + def record_statement( + _connection: Any, + _cursor: Any, + statement: str, + _parameters: Any, + _context: Any, + _executemany: bool, + ) -> None: + if "pg_advisory_xact_lock(" in statement: + operations.append("admission") + elif statement.startswith("INSERT INTO series "): + operations.append("insert") + + event.listen(engine.sync_engine, "before_cursor_execute", record_statement) + try: + yield operations + finally: + event.remove(engine.sync_engine, "before_cursor_execute", record_statement) + + +async def _try_admission(session: AsyncSession) -> bool: + acquired = await session.scalar( + select( + func.pg_try_advisory_xact_lock( + coordination._LOCK_NAMESPACE, coordination._LOCK_RESOURCE + ) + ) + ) + await session.rollback() + return bool(acquired) + + +def _series_count() -> Select[int]: + return select(func.count()).select_from(Series.__table__) + + +async def test_clean_admission_precedes_core_query_autoflush( + identity_probe_db: IdentityProbeDatabase, +) -> None: + engine, factory, _ = identity_probe_db + with _record_admission_and_writes(engine) as operations: + async with factory() as owner, factory() as observer: + assert not owner.new and not owner.dirty and not owner.deleted + await coordination.lock_file_mutation_admission(owner) + assert not await _try_admission(observer), "Admission must exclude another transaction" + pending = Series(title="Clean admission", sort_title="clean admission") + owner.add(pending) + assert (await owner.execute(_series_count())).scalar_one() == 1 + assert pending not in owner.new + assert operations == ["admission", "insert"] + assert not await _try_admission(observer) + await owner.rollback() + assert (await observer.execute(_series_count())).scalar_one() == 0 + assert await _try_admission(observer), "Rollback must release admission" + + +async def test_repeated_admission_autoflushes_only_after_the_mutex_is_already_held( + identity_probe_db: IdentityProbeDatabase, +) -> None: + engine, factory, _ = identity_probe_db + with _record_admission_and_writes(engine) as operations: + async with factory() as owner, factory() as observer: + await coordination.lock_file_mutation_admission(owner) + assert not await _try_admission(observer), "Initial admission must own the mutex" + pending = Series(title="Repeated admission", sort_title="repeated admission") + owner.add(pending) + await coordination.lock_file_mutation_admission(owner) + assert pending not in owner.new, "The Core advisory-lock query must autoflush" + assert operations == ["admission", "insert", "admission"] + assert not await _try_admission(observer) + await owner.rollback() + assert (await observer.execute(_series_count())).scalar_one() == 0 + assert await _try_admission(observer), "Rollback must release repeated admission" diff --git a/tests/unit/test_database_autoflush_admission.py b/tests/unit/test_database_autoflush_admission.py new file mode 100644 index 00000000..9675c299 --- /dev/null +++ b/tests/unit/test_database_autoflush_admission.py @@ -0,0 +1,132 @@ +"""Pending ORM writes must respect maintenance admission for every query form.""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Any, Literal + +import pytest +from sqlalchemy import event, func, select, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + +import pullbox.database as database + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator + from pathlib import Path + + from sqlalchemy import Select + from sqlalchemy.sql.elements import TextClause + + +class AutoflushProbeBase(DeclarativeBase): + pass + + +class AutoflushProbe(AutoflushProbeBase): + __tablename__ = "autoflush_probe" + + id: Mapped[int] = mapped_column(primary_key=True) + + +type StatementKind = Literal["text", "core", "orm"] + + +def _count_statement(kind: StatementKind) -> Select[int] | TextClause: + if kind == "text": + return text("SELECT count(*) FROM autoflush_probe") + if kind == "core": + return select(func.count()).select_from(AutoflushProbe.__table__) + return select(func.count(AutoflushProbe.id)) + + +@pytest.fixture +async def autoflush_database( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> AsyncGenerator[tuple[async_sessionmaker[database.GateAwareAsyncSession], list[str]], None]: + gate = asyncio.Event() + gate.set() + monkeypatch.setattr(database, "_maintenance_gate", gate) + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'autoflush.db'}") + async with engine.begin() as connection: + await connection.run_sync(AutoflushProbeBase.metadata.create_all) + statements: list[str] = [] + + def record_statement( + _connection: Any, + _cursor: Any, + statement: str, + _parameters: Any, + _context: Any, + _executemany: bool, + ) -> None: + statements.append(statement) + + event.listen(engine.sync_engine, "before_cursor_execute", record_statement) + try: + yield async_sessionmaker(engine, class_=database.GateAwareAsyncSession), statements + finally: + gate.set() + event.remove(engine.sync_engine, "before_cursor_execute", record_statement) + await engine.dispose() + + +@pytest.mark.parametrize("statement_kind", ["text", "core", "orm"]) +async def test_autoflush_waits_for_maintenance_and_remains_rollbackable( + autoflush_database: tuple[async_sessionmaker[database.GateAwareAsyncSession], list[str]], + monkeypatch: pytest.MonkeyPatch, + statement_kind: StatementKind, +) -> None: + factory, statements = autoflush_database + admitted = asyncio.Event() + wait_for_ready = database.wait_for_database_ready + + async def observe_admission() -> None: + admitted.set() + await wait_for_ready() + + monkeypatch.setattr(database, "wait_for_database_ready", observe_admission) + async with factory() as session: + pending = AutoflushProbe(id=1) + session.add(pending) + assert pending in session.new + database._maintenance_gate.clear() + read_task = asyncio.create_task(session.execute(_count_statement(statement_kind))) + admission_task = asyncio.create_task(admitted.wait()) + try: + completed, _ = await asyncio.wait( + (read_task, admission_task), timeout=2, return_when=asyncio.FIRST_COMPLETED + ) + assert admission_task in completed, "Query bypassed maintenance admission" + assert not read_task.done(), "Query completed while maintenance was active" + assert statements == [], "Autoflush issued SQL before maintenance admission" + database._maintenance_gate.set() + result = await asyncio.wait_for(read_task, 2) + assert result.scalar_one() == 1, "All query forms must autoflush pending ORM writes" + assert pending not in session.new + await session.rollback() + assert (await session.execute(_count_statement("core"))).scalar_one() == 0 + finally: + database._maintenance_gate.set() + admission_task.cancel() + read_task.cancel() + await asyncio.gather(read_task, admission_task, return_exceptions=True) + + +@pytest.mark.parametrize("statement_kind", ["text", "core", "orm"]) +async def test_explicit_no_autoflush_preserves_pending_writes_until_an_orm_query( + autoflush_database: tuple[async_sessionmaker[database.GateAwareAsyncSession], list[str]], + statement_kind: StatementKind, +) -> None: + factory, _ = autoflush_database + async with factory() as session: + pending = AutoflushProbe(id=1) + session.add(pending) + with session.no_autoflush: + assert (await session.execute(_count_statement(statement_kind))).scalar_one() == 0 + assert pending in session.new + assert (await session.execute(_count_statement("orm"))).scalar_one() == 1 + assert pending not in session.new + await session.rollback() + assert (await session.execute(_count_statement("core"))).scalar_one() == 0 diff --git a/tests/unit/test_dependency_constraints.py b/tests/unit/test_dependency_constraints.py index b335f292..8f5884ab 100644 --- a/tests/unit/test_dependency_constraints.py +++ b/tests/unit/test_dependency_constraints.py @@ -6,7 +6,7 @@ from packaging.requirements import Requirement -def test_sqlalchemy_dependency_stays_on_qualified_20_line() -> None: +def test_sqlalchemy_dependency_stays_on_qualified_21_line() -> None: project = tomllib.loads(Path("pyproject.toml").read_text()) sqlalchemy = next( requirement @@ -14,6 +14,11 @@ def test_sqlalchemy_dependency_stays_on_qualified_20_line() -> None: if (requirement := Requirement(dependency)).name == "sqlalchemy" ) - assert "2.0.53" in sqlalchemy.specifier + assert "asyncio" in sqlalchemy.extras + assert "2.0.54" not in sqlalchemy.specifier assert "2.1.1" not in sqlalchemy.specifier + assert "2.1.3" not in sqlalchemy.specifier + assert "2.1.4" in sqlalchemy.specifier + assert "2.1.5" in sqlalchemy.specifier + assert "2.2.0" not in sqlalchemy.specifier assert "3.0.0" not in sqlalchemy.specifier diff --git a/tests/unit/test_import_database_sessions.py b/tests/unit/test_import_database_sessions.py new file mode 100644 index 00000000..97e151b0 --- /dev/null +++ b/tests/unit/test_import_database_sessions.py @@ -0,0 +1,211 @@ +"""Import session selection follows SQLite URL semantics, not rendered spelling.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace +from typing import TYPE_CHECKING, cast + +import pytest +from sqlalchemy import URL, event, literal, select +from sqlalchemy.exc import SAWarning +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.orm import ORMExecuteState, Session + +from pullbox.core.exceptions import JobPausedError +from pullbox.services.import_file_matching import _spool_grouped_cohort_rows +from pullbox.services.import_job_controls import raise_if_job_cancelled_immediately +from pullbox.services.import_job_execution_progress import progress_session_factory_for_runtime +from pullbox.services.import_provider_cache import is_in_memory_sqlite_engine + +if TYPE_CHECKING: + from pathlib import Path + + from sqlalchemy.ext.asyncio import AsyncEngine + + +@pytest.mark.parametrize( + ("driver", "database", "query", "expected"), + [ + ("sqlite+aiosqlite", None, {}, True), + ("sqlite+aiosqlite", "", {}, True), + ("sqlite+aiosqlite", ":memory:", {}, True), + ("sqlite+aiosqlite", "file:import-probe", {"mode": "memory", "uri": "true"}, True), + ("sqlite+aiosqlite", "file:import-probe", {"mode": "memory", "uri": " YES "}, True), + ("sqlite+aiosqlite", "file::memory:", {"uri": "1"}, True), + ("sqlite+aiosqlite", "file:import-probe", {"mode": "memory"}, False), + ("sqlite+aiosqlite", "file:import-probe", {"mode": "memory", "uri": "false"}, False), + ("sqlite+aiosqlite", "file::memory:", {}, False), + ("sqlite+aiosqlite", "file:import-probe", {"mode": "ro", "uri": "true"}, False), + ("sqlite+aiosqlite", "file:import-probe", {"mode": "rw", "uri": "true"}, False), + ("sqlite+aiosqlite", "file:import-probe", {"mode": "rwc", "uri": "true"}, False), + ("sqlite+aiosqlite", "file:import-probe", {"note": "mode=memory", "uri": "true"}, False), + ("sqlite+aiosqlite", "local:memory:.db", {}, False), + ("sqlite+aiosqlite", "mode=memory.db", {}, False), + ("postgresql+asyncpg", ":memory:", {}, False), + ("postgresql+asyncpg", "file:import-probe", {"mode": "memory", "uri": "true"}, False), + ], +) +async def test_memory_classifier_uses_structured_url( + driver: str, database: str | None, query: dict[str, str], expected: bool +) -> None: + url = URL.create(driver, database=database, query=query) + if driver == "postgresql+asyncpg": + # Backend classification needs the real URL, not the optional database driver. + engine = cast("AsyncEngine", SimpleNamespace(sync_engine=SimpleNamespace(url=url))) + elif ( + driver == "sqlite+aiosqlite" + and query.get("mode") == "memory" + and query.get("uri") + in { + None, + "false", + } + ): + with pytest.warns(SAWarning, match="SQLite URI arguments require"): + engine = create_async_engine(url) + else: + engine = create_async_engine(url) + try: + assert is_in_memory_sqlite_engine(engine) is expected + finally: + if driver == "sqlite+aiosqlite": + await engine.dispose() + + +def _test_url(kind: str, tmp_path: Path) -> URL: + if kind == "memory": + return URL.create("sqlite+aiosqlite", database=":memory:") + if kind == "empty": + return URL.create("sqlite+aiosqlite") + if kind == "uri-memory": + return URL.create( + "sqlite+aiosqlite", + database=f"file:{tmp_path / 'named-memory'}", + query={"mode": "memory", "cache": "shared", "uri": "true"}, + ) + filename = "local:memory:.db" if kind == "file-colon" else "local.db" + return URL.create("sqlite+aiosqlite", database=str(tmp_path / filename)) + + +@pytest.mark.parametrize("kind", ["memory", "empty", "uri-memory", "file", "file-colon"]) +async def test_control_sessions_preserve_memory_or_file_isolation( + kind: str, tmp_path: Path +) -> None: + engine = create_async_engine(_test_url(kind, tmp_path)) + memory = kind in {"memory", "empty", "uri-memory"} + seen: list[AsyncSession] = [] + try: + async with AsyncSession(engine, expire_on_commit=False) as active: + connection = await active.connection() + original_dbapi = connection.sync_connection.connection.dbapi_connection + + async def check_cancel(session: AsyncSession, job_id: int) -> None: + seen.append(session) + assert job_id == 37 + assert (session is active) is memory + control_connection = await session.connection() + assert ( + control_connection.sync_connection.connection.dbapi_connection is original_dbapi + ) is memory + assert await session.scalar(select(literal(17))) == 17 + if memory: + assert session.autoflush is False + raise JobPausedError("paused") + + with pytest.raises(JobPausedError, match="paused"): + await raise_if_job_cancelled_immediately( + active, 37, raise_if_cancelled=check_cancel + ) + assert len(seen) == 1 + assert active.autoflush is True + assert await active.scalar(select(literal(19))) == 19 + finally: + await engine.dispose() + + +@pytest.mark.parametrize("kind", ["memory", "empty", "uri-memory", "file", "file-colon"]) +async def test_progress_sessions_preserve_memory_or_file_isolation( + kind: str, tmp_path: Path +) -> None: + engine = create_async_engine(_test_url(kind, tmp_path)) + memory = kind in {"memory", "empty", "uri-memory"} + try: + async with AsyncSession(engine, expire_on_commit=False) as active: + connection = await active.connection() + original_dbapi = connection.sync_connection.connection.dbapi_connection + progress_factory = progress_session_factory_for_runtime(active) + if memory: + assert progress_factory is None + else: + assert progress_factory is not None + async with progress_factory() as progress: + assert progress is not active + assert type(progress) is type(active) + progress_connection = await progress.connection() + assert ( + progress_connection.sync_connection.connection.dbapi_connection + is not original_dbapi + ) + assert await progress.scalar(select(literal(23))) == 23 + finally: + await engine.dispose() + + +async def test_connection_bound_memory_session_keeps_active_control_and_progress( + tmp_path: Path, +) -> None: + engine = create_async_engine(_test_url("memory", tmp_path)) + try: + async with engine.connect() as connection, AsyncSession(connection) as active: + + async def check_cancel(session: AsyncSession, _job_id: int) -> None: + assert session is active + assert await session.scalar(select(literal(29))) == 29 + + await raise_if_job_cancelled_immediately(active, 37, raise_if_cancelled=check_cancel) + assert progress_session_factory_for_runtime(active) is None + finally: + await engine.dispose() + + +@pytest.mark.parametrize("kind", ["memory", "empty", "uri-memory", "file", "file-colon"]) +async def test_grouped_cohort_scan_reuses_memory_session_or_isolates_file_reader( + kind: str, tmp_path: Path +) -> None: + engine = create_async_engine(_test_url(kind, tmp_path)) + memory = kind in {"memory", "empty", "uri-memory"} + observed: list[Session] = [] + cancelled_checks: list[int] = [] + + def record_scan(state: ORMExecuteState) -> None: + if state.execution_options.get("pullbox_cohort_scan") == "url-regression": + observed.append(state.session) + + event.listen(Session, "do_orm_execute", record_scan) + try: + async with async_sessionmaker(engine)() as active: + + async def check_cancel(session: AsyncSession, job_id: int) -> None: + assert session is active + assert job_id == 37 + cancelled_checks.append(job_id) + + spool = await _spool_grouped_cohort_rows( + active, + job_id=37, + query=select(literal(1)).union_all(select(literal(2))), + batch_size=1, + scan_name="url-regression", + row_payload=lambda row: [row[0]], + raise_if_cancelled=check_cancel, + ) + with spool: + assert [json.loads(line) for line in spool] == [[1], [2]] + assert len(observed) == 1 + assert (observed[0] is active.sync_session) is memory + assert cancelled_checks == [37, 37, 37, 37] + assert await active.scalar(select(literal(31))) == 31 + finally: + event.remove(Session, "do_orm_execute", record_scan) + await engine.dispose() diff --git a/tests/unit/test_scheduler_stats_persistence.py b/tests/unit/test_scheduler_stats_persistence.py index 5f5703de..bee7ba00 100644 --- a/tests/unit/test_scheduler_stats_persistence.py +++ b/tests/unit/test_scheduler_stats_persistence.py @@ -3,12 +3,15 @@ from __future__ import annotations from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace +from typing import TYPE_CHECKING, cast from unittest.mock import AsyncMock, MagicMock import pytest -from sqlalchemy import select +from sqlalchemy import URL, select from sqlalchemy.exc import OperationalError -from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from pullbox.core.scheduler_stats import TaskStats from pullbox.core.scheduler_stats_persistence import ( @@ -20,6 +23,9 @@ ) from pullbox.models.scheduler_task_stat import ScheduledTaskStat +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncEngine + @pytest.mark.asyncio async def test_upsert_task_stat_row_round_trips_all_event_fields(async_engine) -> None: @@ -134,3 +140,67 @@ def test_resolve_legacy_task_stats_sidecar_path_uses_sqlite_db_neighbor() -> Non assert str(sidecar) == "/config/pullbox/scheduled_task_stats.json" assert resolve_legacy_task_stats_sidecar_path("sqlite+aiosqlite:///:memory:") is None assert resolve_legacy_task_stats_sidecar_path("postgresql://example") is None + + +@pytest.mark.parametrize( + ("driver", "database", "expected_directory"), + [ + ("sqlite+aiosqlite", ":memory:", None), + ("sqlite+aiosqlite", None, None), + ("sqlite+aiosqlite", "", None), + ("sqlite+aiosqlite", "/tmp/ordinary/pullbox.db", "/tmp/ordinary"), + ("sqlite+aiosqlite", "/tmp/library:one/pullbox.db", "/tmp/library:one"), + ("sqlite+aiosqlite", "/tmp/library:memory:/pullbox.db", "/tmp/library:memory:"), + ("sqlite+aiosqlite", "/tmp/library?one/pullbox.db", "/tmp/library?one"), + ("sqlite+aiosqlite", "/tmp/library%3Aone/pullbox.db", "/tmp/library%3Aone"), + ("postgresql+asyncpg", "sqlite:memory:", None), + ], +) +async def test_legacy_sidecar_uses_structured_runtime_database( + driver: str, database: str | None, expected_directory: str | None +) -> None: + """A rendered engine URL must not become an encoded or imaginary file path.""" + url = URL.create(driver, database=database) + if driver == "postgresql+asyncpg": + # URL resolution does not require a connection or the optional database driver. + engine = cast("AsyncEngine", SimpleNamespace(url=url)) + else: + engine = create_async_engine(url) + try: + factory = async_sessionmaker(engine, expire_on_commit=False) + runtime_url = resolve_runtime_db_url("sqlite+aiosqlite:////unused/default.db", factory) + expected = ( + Path(expected_directory) / "scheduled_task_stats.json" + if expected_directory is not None + else None + ) + assert resolve_legacy_task_stats_sidecar_path(runtime_url) == expected + finally: + if driver == "sqlite+aiosqlite": + await engine.dispose() + + +@pytest.mark.parametrize("db_url", ["", "not-a-sqlite-url", "https://sqlite.example/database"]) +def test_legacy_sidecar_rejects_non_database_urls(db_url: str) -> None: + assert resolve_legacy_task_stats_sidecar_path(db_url) is None + + +@pytest.mark.parametrize( + ("database", "query", "is_memory"), + [ + ("file::memory:", {"uri": "true"}, True), + ("file:sidecar", {"uri": "true", "mode": "memory"}, True), + ("file:sidecar", {"uri": " YES ", "mode": "memory"}, True), + ("file::memory:", {"uri": "t"}, True), + ("file:sidecar", {"uri": "false", "mode": "memory"}, False), + ("file:sidecar", {"uri": "true", "mode": "rwc"}, False), + ("file:sidecar", {"uri": "true", "cache": "mode=memory"}, False), + ("file:sidecar", {"mode": "memory"}, False), + ], +) +def test_legacy_sidecar_classifies_uri_memory_without_query_substrings( + database: str, query: dict[str, str], is_memory: bool +) -> None: + url = URL.create("sqlite+aiosqlite", database=database, query=query) + expected = None if is_memory else Path("scheduled_task_stats.json") + assert resolve_legacy_task_stats_sidecar_path(str(url)) == expected