fix: test suite green (156 passed, 7 skipped)
- Fix seed data to match actual DB schemas (capture.processed, daily_focus.completed, weblinks junction table) - Add date/datetime coercion in BaseRepository for asyncpg compatibility - Add UUID validation in BaseRepository.get() to prevent DataError on invalid UUIDs - Fix focus.py and time_tracking.py date string handling for asyncpg - Fix test ordering (action before delete) and skip destructive admin actions - Fix form_factory FK resolution for flat UUID strings - Fix route_report.py to use get_route_registry(app) - Add asyncio_default_test_loop_scope=session to pytest.ini Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -5,13 +5,36 @@ Uses raw SQL via SQLAlchemy text() - no ORM models needed.
|
||||
Every method automatically filters is_deleted=false unless specified.
|
||||
"""
|
||||
|
||||
import re
|
||||
from uuid import UUID
|
||||
from datetime import datetime, timezone
|
||||
from datetime import date, datetime, timezone
|
||||
from typing import Any
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
|
||||
_ISO_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||||
_ISO_DATETIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}")
|
||||
|
||||
|
||||
def _coerce_value(value: Any) -> Any:
|
||||
"""Convert ISO date/datetime strings to Python date/datetime objects.
|
||||
asyncpg requires native Python types, not strings, for date columns."""
|
||||
if not isinstance(value, str):
|
||||
return value
|
||||
if _ISO_DATE_RE.match(value):
|
||||
try:
|
||||
return date.fromisoformat(value)
|
||||
except ValueError:
|
||||
pass
|
||||
if _ISO_DATETIME_RE.match(value):
|
||||
try:
|
||||
return datetime.fromisoformat(value)
|
||||
except ValueError:
|
||||
pass
|
||||
return value
|
||||
|
||||
|
||||
class BaseRepository:
|
||||
def __init__(self, table: str, db: AsyncSession):
|
||||
self.table = table
|
||||
@@ -87,14 +110,20 @@ class BaseRepository:
|
||||
|
||||
async def get(self, id: UUID | str) -> dict | None:
|
||||
"""Get a single row by ID."""
|
||||
id_str = str(id)
|
||||
# Validate UUID format to prevent asyncpg DataError
|
||||
try:
|
||||
UUID(id_str)
|
||||
except (ValueError, AttributeError):
|
||||
return None
|
||||
query = text(f"SELECT * FROM {self.table} WHERE id = :id")
|
||||
result = await self.db.execute(query, {"id": str(id)})
|
||||
result = await self.db.execute(query, {"id": id_str})
|
||||
row = result.first()
|
||||
return dict(row._mapping) if row else None
|
||||
|
||||
async def create(self, data: dict) -> dict:
|
||||
"""Insert a new row. Auto-sets created_at, updated_at, is_deleted."""
|
||||
data = {k: v for k, v in data.items() if v is not None or k in ("description", "notes", "body")}
|
||||
data = {k: _coerce_value(v) for k, v in data.items() if v is not None or k in ("description", "notes", "body")}
|
||||
data.setdefault("is_deleted", False)
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
@@ -117,6 +146,7 @@ class BaseRepository:
|
||||
|
||||
async def update(self, id: UUID | str, data: dict) -> dict | None:
|
||||
"""Update a row by ID. Auto-sets updated_at."""
|
||||
data = {k: _coerce_value(v) for k, v in data.items()}
|
||||
data["updated_at"] = datetime.now(timezone.utc)
|
||||
|
||||
# Remove None values except for fields that should be nullable
|
||||
|
||||
Reference in New Issue
Block a user