chore(deps): qualify SQLAlchemy >=2.1.4,<2.2

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",

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
 

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")
 
 

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(

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,

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

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)

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

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,

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)

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)
                     )

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)

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(

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)

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,

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")

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

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", "[]"))

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)