chore(backend): clear 102 ruff lint errors

Config: drop TCH from select (breaks FastAPI runtime get_type_hints on
route signatures) and allowlist fastapi.Depends/Query/etc. for B008 so
DI defaults don't fire. Auto-fix I001/F401/UP017/F541 across 15 files.
Manual: B023 capture `remaining` via default arg in stac.py score() to
pin the closure; SIM102 collapse nested if in demographics; B905
strict=False on zip() in census + imagery; SIM117 combine nested
`with patch()` pairs in test_parcels.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: log0s

backend/app/api/v1/events.py

@@ -20,7 +20,7 @@
     PropertyEventsResponse,
 )
 from app.services import property_events as property_events_service
-from app.services.county_adapters import get_adapter_for_county, get_supported_counties
+from app.services.county_adapters import get_adapter_for_county
 
 router = APIRouter()
 

backend/app/api/v1/geocode.py

@@ -7,7 +7,6 @@
 
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Query
-from fastapi.responses import JSONResponse
 from redis.exceptions import RedisError
 from sqlalchemy.orm import Session
 

backend/app/api/v1/imagery.py

@@ -16,7 +16,8 @@
 from datetime import date
 
 import httpx
-from fastapi import APIRouter, Depends, HTTPException, Query, Response as FastAPIResponse
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import Response as FastAPIResponse
 from fastapi.responses import Response
 from redis.exceptions import RedisError
 from sqlalchemy.orm import Session
@@ -171,7 +172,7 @@ async def list_imagery(
         return_exceptions=True,
     )
     signed_map: dict[str, str] = {
-        u: (r if isinstance(r, str) else u) for u, r in zip(url_list, results)
+        u: (r if isinstance(r, str) else u) for u, r in zip(url_list, results, strict=False)
     }
 
     snapshot_responses: list[ImagerySnapshotResponse] = []

backend/app/services/census.py

@@ -169,14 +169,14 @@ async def _request(
 
         if resp.status_code in (204, 404):
             logger.info(
-                f"Census API: no data for tract",
+                "Census API: no data for tract",
                 extra={"url": url, "status": resp.status_code},
             )
             return None
 
         if resp.status_code != 200:
             logger.error(
-                f"Census API error",
+                "Census API error",
                 extra={"url": url, "status": resp.status_code, "body": resp.text[:500]},
             )
             raise CensusApiError(
@@ -213,7 +213,7 @@ def _parse_response(data: list[list[str]]) -> dict[str, int | float | None]:
 
     return {
         h: _to_number(v)
-        for h, v in zip(headers, values)
+        for h, v in zip(headers, values, strict=False)
         if h not in geo_fields
     }
 

backend/app/services/county_adapters.py

@@ -15,7 +15,7 @@
 import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from datetime import date
+from datetime import UTC, date
 from typing import Any
 
 from app.services.arcgis import query_feature_service
@@ -720,9 +720,9 @@ def _parse_epoch_ms(value: Any) -> date | None:
     if value is None:
         return None
     try:
-        from datetime import datetime, timezone
+        from datetime import datetime
 
-        return datetime.fromtimestamp(int(value) / 1000, tz=timezone.utc).date()
+        return datetime.fromtimestamp(int(value) / 1000, tz=UTC).date()
     except (ValueError, TypeError, OSError):
         return parse_date(str(value))
 

backend/app/services/demographics.py

@@ -60,9 +60,8 @@ def upsert_census_snapshot(
     vacancy_rate: float | None = None
     total_housing = data.get("total_housing_units")
     occupied = data.get("occupied_housing_units")
-    if total_housing and total_housing > 0:
-        if occupied is not None:
-            vacancy_rate = round((total_housing - occupied) / total_housing, 4)
+    if total_housing and total_housing > 0 and occupied is not None:
+        vacancy_rate = round((total_housing - occupied) / total_housing, 4)
 
     snap_id = uuid.uuid4()
 

backend/app/services/imagery.py

@@ -13,9 +13,10 @@
 import logging
 import uuid
 from dataclasses import dataclass
-from datetime import date, datetime, timezone
+from datetime import UTC, date, datetime
 
-from sqlalchemy import select, text as sa_text
+from sqlalchemy import select
+from sqlalchemy import text as sa_text
 from sqlalchemy.orm import Session
 
 from app.models.parcels import Parcel, TimelineRequest, TimelineRequestTask
@@ -137,9 +138,9 @@ def update_request_task(
     task.status = status
     task.items_found = items_found
     if status == "processing":
-        task.started_at = datetime.now(tz=timezone.utc)
+        task.started_at = datetime.now(tz=UTC)
     elif status in ("complete", "failed", "skipped"):
-        task.completed_at = datetime.now(tz=timezone.utc)
+        task.completed_at = datetime.now(tz=UTC)
     if error_message:
         task.error_message = error_message
     db.commit()
@@ -154,7 +155,7 @@ def update_timeline_request_status(
     """Update the parent timeline request status."""
     request.status = status
     if status in ("complete", "failed"):
-        request.completed_at = datetime.now(tz=timezone.utc)
+        request.completed_at = datetime.now(tz=UTC)
     if error_message:
         request.error_message = error_message
     db.commit()

backend/app/services/stac.py

@@ -318,15 +318,18 @@ def select_naip_items(
         candidates = list(year_items)
 
         while candidates and len(selected_for_year) < max_tiles_per_year:
-            def score(item: dict[str, object]) -> tuple[float, float]:
+            def score(
+                item: dict[str, object],
+                _remaining: tuple[float, float, float, float] = remaining,
+            ) -> tuple[float, float]:
                 bbox = item.get("bbox")
                 if not bbox or len(bbox) < 4:  # type: ignore[arg-type]
                     return (0.0, float(abs(_doy(item) - target_doy)))
                 ib = (
                     float(bbox[0]), float(bbox[1]),  # type: ignore[index]
                     float(bbox[2]), float(bbox[3]),  # type: ignore[index]
                 )
-                area = _bbox_intersection_area(ib, remaining)
+                area = _bbox_intersection_area(ib, _remaining)
                 # Maximize area, minimize doy distance
                 return (-area, float(abs(_doy(item) - target_doy)))
 

backend/app/tasks/timeline.py

@@ -15,19 +15,19 @@
 
 import httpx
 
-from app.services import stac as stac_service
-from app.services import imagery as imagery_service
 from app.services import demographics as demographics_service
+from app.services import imagery as imagery_service
 from app.services import property_events as property_events_service
+from app.services import stac as stac_service
 from app.services.address_normalizer import extract_search_terms, is_address_match
-from app.services.county_adapters import get_adapter_for_county
 from app.services.census import (
     ACS5_YEARS,
     DECENNIAL_YEARS,
     CensusApiError,
     CensusFetcher,
     parse_tract_fips,
 )
+from app.services.county_adapters import get_adapter_for_county
 from app.tasks.celery_app import celery_app
 
 logger = logging.getLogger(__name__)
@@ -148,6 +148,7 @@ async def _fetch_source(
     with SessionLocal() as db:
         # Find and update the task row
         from sqlalchemy import select as sa_select
+
         from app.models.parcels import TimelineRequest, TimelineRequestTask
 
         request = db.execute(
@@ -215,9 +216,10 @@ async def _fetch_source(
             extra={"error": str(exc)},
         )
         with SessionLocal() as db:
-            from app.models.parcels import TimelineRequestTask
             from sqlalchemy import select as sa_select
 
+            from app.models.parcels import TimelineRequestTask
+
             task_row = db.execute(
                 sa_select(TimelineRequestTask)
                 .where(TimelineRequestTask.timeline_request_id == timeline_request_id)
@@ -300,9 +302,10 @@ async def _fetch_source(
                 items_saved += 1
 
         # Update task status
-        from app.models.parcels import TimelineRequestTask
         from sqlalchemy import select as sa_select
 
+        from app.models.parcels import TimelineRequestTask
+
         task_row = db.execute(
             sa_select(TimelineRequestTask)
             .where(TimelineRequestTask.timeline_request_id == timeline_request_id)
@@ -331,9 +334,10 @@ async def _fetch_census(
 
     Returns the number of census snapshots saved.
     """
+    from sqlalchemy import select as sa_select
+
     from app.db import SessionLocal
     from app.models.parcels import TimelineRequestTask
-    from sqlalchemy import select as sa_select
 
     try:
         state_fips, county_fips, tract_code = parse_tract_fips(tract_fips)
@@ -427,7 +431,7 @@ async def _fetch_census(
                 db, task_row, "complete", items_found=items_saved
             )
 
-    logger.info(f"Census fetch complete", extra={"items_saved": items_saved, "tract": tract_fips})
+    logger.info("Census fetch complete", extra={"items_saved": items_saved, "tract": tract_fips})
     return items_saved
 
 
@@ -442,9 +446,10 @@ async def _fetch_property(
 
     Returns the number of events saved.
     """
+    from sqlalchemy import select as sa_select
+
     from app.db import SessionLocal
     from app.models.parcels import TimelineRequestTask
-    from sqlalchemy import select as sa_select
 
     adapter = get_adapter_for_county(county)
 
@@ -587,9 +592,10 @@ async def _fetch_property(
 
 async def _run_timeline(timeline_request_id: str) -> dict[str, Any]:
     """Orchestrate all imagery sources for a timeline request."""
+    from sqlalchemy import select as sa_select
+
     from app.db import SessionLocal
     from app.models.parcels import TimelineRequest
-    from sqlalchemy import select as sa_select
 
     req_uuid = uuid.UUID(timeline_request_id)
 

backend/pyproject.toml

@@ -54,9 +54,24 @@ line-length = 100
 target-version = "py312"
 
 [tool.ruff.lint]
-select = ["E", "F", "I", "UP", "B", "SIM", "TCH"]
+select = ["E", "F", "I", "UP", "B", "SIM"]
 ignore = ["E501"]
 
+[tool.ruff.lint.flake8-bugbear]
+# FastAPI dependency-injection idiom: Depends(), Query(), etc. are required
+# as argument defaults. Allowlist them so B008 doesn't fire on route signatures.
+extend-immutable-calls = [
+    "fastapi.Depends",
+    "fastapi.Query",
+    "fastapi.Path",
+    "fastapi.Header",
+    "fastapi.Cookie",
+    "fastapi.Body",
+    "fastapi.Form",
+    "fastapi.File",
+    "fastapi.Security",
+]
+
 [tool.mypy]
 python_version = "3.12"
 strict = true