Tech-debt batch 4: sqlmodel col() sweep, shared user-directory SQL helpers (#591 #596)

framework/db/simple_module_db/user_directory_sql.py

@@ -0,0 +1,41 @@
+"""SQL-backed user directory helpers shared by account providers."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from simple_module_core.user_directory import DirectoryUser
+from sqlalchemy import or_
+from sqlalchemy.sql.elements import ColumnElement
+
+from simple_module_db.search import LIKE_ESCAPE_CHAR, like_contains_pattern
+
+
+def name_or_email_contains(full_name: Any, email: Any, term: str) -> ColumnElement[bool]:
+    """Match a literal substring in either provider column, ignoring case."""
+    pattern = like_contains_pattern(term)
+    return or_(
+        full_name.ilike(pattern, escape=LIKE_ESCAPE_CHAR),
+        email.ilike(pattern, escape=LIKE_ESCAPE_CHAR),
+    )
+
+
+def to_directory_user(
+    row: Any,
+    key: str,
+    *,
+    roles: tuple[str, ...] | None = None,
+    is_active: bool = True,
+) -> DirectoryUser:
+    """Convert identity columns, preserving unknown versus empty roles.
+
+    Providers supply account availability and roles when known. ``None`` means
+    unknown roles; an empty tuple means no assigned roles.
+    """
+    return DirectoryUser(
+        id=key,
+        email=row.email,
+        full_name=row.full_name,
+        is_active=is_active,
+        roles=roles,
+    )

framework/db/tests/test_db_user_directory_sql.py

@@ -0,0 +1,29 @@
+"""Shared directory predicates treat SQL LIKE metacharacters as literal text."""
+
+from types import SimpleNamespace
+
+import pytest
+from simple_module_db.user_directory_sql import name_or_email_contains, to_directory_user
+from sqlalchemy import literal, select
+from sqlmodel import Session, create_engine
+
+
+@pytest.mark.parametrize("term", ["%", "_", "\\"])
+def test_name_or_email_contains_escapes_literal_metacharacters(term: str):
+    with Session(create_engine("sqlite://")) as db:
+        for full_name, email in [(f"Ada{term}Lovelace", "other"), (None, f"ada{term}@test")]:
+            predicate = name_or_email_contains(literal(full_name), literal(email), term)
+            assert db.exec(select(predicate)).scalar_one()
+        predicate = name_or_email_contains(literal("Ada Lovelace"), literal("ada@test"), term)
+        assert not db.exec(select(predicate)).scalar_one()
+
+
+@pytest.mark.parametrize("roles", [None, (), ("editor",)])
+def test_conversion_preserves_known_and_unknown_roles(roles: tuple[str, ...] | None):
+    row = SimpleNamespace(email="ada@test", full_name=None)
+    user = to_directory_user(row, "caller-id", roles=roles, is_active=False)
+    assert user.id == "caller-id"
+    assert user.email == row.email
+    assert user.full_name is None
+    assert user.roles == roles
+    assert not user.is_active

modules/api_tokens/api_tokens/resolver.py

@@ -24,6 +24,7 @@
 from simple_module_core.user_directory import UserDirectory, get_user_directory
 from simple_module_db import ALL_TENANTS_OPTION, independent_session
 from sqlalchemy import or_, select, update
+from sqlmodel import col
 from starlette.requests import Request
 
 from api_tokens.constants import (
@@ -112,7 +113,7 @@ async def touch_last_used(app: Any, token: ApiToken, now: datetime) -> None:
     stmt = (
         update(ApiToken)
         .where(
-            ApiToken.id == token.id,  # ty: ignore[invalid-argument-type]
+            col(ApiToken.id) == token.id,
             or_(ApiToken.last_used_at.is_(None), ApiToken.last_used_at <= cutoff),
         )
         .values(last_used_at=now, updated_at=ApiToken.updated_at)
@@ -136,7 +137,7 @@ async def resolve_api_token(request: Request) -> UserContext | None:
         async with app.state.sm.db.session_factory() as db:
             stmt = (
                 select(ApiToken)
-                .where(ApiToken.token_hash == hash_token(credential))  # ty: ignore[invalid-argument-type]
+                .where(col(ApiToken.token_hash) == hash_token(credential))
                 .execution_options(**{ALL_TENANTS_OPTION: True})
             )
             token = (await db.execute(stmt)).scalar_one_or_none()

modules/api_tokens/api_tokens/service.py

@@ -17,6 +17,7 @@
 from simple_module_db import ALL_TENANTS_OPTION, all_tenants
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel import col
 
 from api_tokens.constants import FORBIDDEN_SERVICE_ROLES, TokenKind
 from api_tokens.contracts.schemas import TokenView
@@ -115,7 +116,10 @@ async def list_for_owner(self, owner_user_id: uuid.UUID) -> list[ApiToken]:
         """
         stmt = (
             select(ApiToken)
-            .where(ApiToken.owner_user_id == owner_user_id, ApiToken.kind == TokenKind.PERSONAL)  # ty: ignore[invalid-argument-type]
+            .where(
+                col(ApiToken.owner_user_id) == owner_user_id,
+                col(ApiToken.kind) == TokenKind.PERSONAL,
+            )
             .order_by(ApiToken.created_at.desc())
             .execution_options(**{ALL_TENANTS_OPTION: True})
         )
@@ -130,16 +134,16 @@ async def get_owned(self, token_id: uuid.UUID, owner_user_id: uuid.UUID) -> ApiT
         stmt = (
             select(ApiToken)
             .where(
-                ApiToken.id == token_id,  # ty: ignore[invalid-argument-type]
-                ApiToken.owner_user_id == owner_user_id,  # ty: ignore[invalid-argument-type]
-                ApiToken.kind == TokenKind.PERSONAL,  # ty: ignore[invalid-argument-type]
+                col(ApiToken.id) == token_id,
+                col(ApiToken.owner_user_id) == owner_user_id,
+                col(ApiToken.kind) == TokenKind.PERSONAL,
             )
             .execution_options(**{ALL_TENANTS_OPTION: True})
         )
         return (await self.db.execute(stmt)).scalar_one_or_none()
 
     async def get_in_tenant(self, token_id: uuid.UUID) -> ApiToken | None:
-        stmt = select(ApiToken).where(ApiToken.id == token_id)  # ty: ignore[invalid-argument-type]
+        stmt = select(ApiToken).where(col(ApiToken.id) == token_id)
         return (await self.db.execute(stmt)).scalar_one_or_none()
 
     async def revoke(self, token: ApiToken, *, cross_tenant: bool = False) -> ApiToken:

modules/audit_log/audit_log/service.py

@@ -8,6 +8,7 @@
 from simple_module_db.provider import DatabaseProvider
 from sqlalchemy import Select, and_, func, literal_column, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel import col
 
 from audit_log.constants import DEFAULT_PAGE_SIZE, MAX_PAGE, MAX_PAGE_SIZE
 from audit_log.contracts.schemas import AuditEntryList, AuditEntryRead
@@ -158,8 +159,10 @@ async def iter_entries(
                 # is what stops a batch boundary swallowing the rest of them.
                 stmt = stmt.where(
                     or_(
-                        AuditEntry.created_at < last_created,  # ty: ignore[invalid-argument-type]
-                        and_(AuditEntry.created_at == last_created, AuditEntry.id > last_id),  # ty: ignore[invalid-argument-type]
+                        col(AuditEntry.created_at) < last_created,
+                        and_(
+                            col(AuditEntry.created_at) == last_created, col(AuditEntry.id) > last_id
+                        ),
                     )
                 )
             stmt = stmt.order_by(AuditEntry.created_at.desc(), AuditEntry.id).limit(batch_size)

modules/background_tasks/background_tasks/_signal_support.py

@@ -15,6 +15,7 @@
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
+from sqlmodel import col
 
 from background_tasks.constants import INTERNAL_TASK_SWEEP_OUTBOX, TaskStatus
 from background_tasks.models import TaskExecution
@@ -53,7 +54,7 @@ def upsert_by_celery_id(
     """
     row: TaskExecution | None = None
     if celery_task_id:
-        stmt = select(TaskExecution).where(TaskExecution.celery_task_id == celery_task_id)  # ty: ignore[invalid-argument-type]
+        stmt = select(TaskExecution).where(col(TaskExecution.celery_task_id) == celery_task_id)
         row = session.execute(stmt).scalar_one_or_none()
     if row is None:
         row = TaskExecution(celery_task_id=celery_task_id, **defaults)

modules/background_tasks/background_tasks/endpoints/api_status.py

@@ -18,6 +18,7 @@
 from simple_module_hosting.permissions import ensure_resolved_permissions
 from simple_module_hosting.token_guard import Authenticated
 from sqlalchemy import select
+from sqlmodel import col
 
 from background_tasks.constants import MODULE_NAME, PERM_VIEW, STATUS_ROUTER_PREFIX
 from background_tasks.contracts.schemas import TaskStatusDTO
@@ -56,7 +57,7 @@ async def get_task_status(
     row = (
         await db.execute(
             select(TaskExecution)
-            .where(TaskExecution.celery_task_id == celery_task_id)  # ty: ignore[invalid-argument-type]
+            .where(col(TaskExecution.celery_task_id) == celery_task_id)
             .order_by(TaskExecution.queued_at.desc())
             .limit(1)
         )

modules/background_tasks/background_tasks/filters.py

@@ -10,6 +10,7 @@
 
 from simple_module_db import LIKE_ESCAPE_CHAR, like_contains_pattern
 from sqlalchemy import ColumnElement, select
+from sqlmodel import col
 
 from background_tasks.constants import PLATFORM_TENANT_FILTER, RETRYABLE_STATUSES, TaskStatus
 from background_tasks.models import TaskExecution
@@ -34,15 +35,15 @@ def execution_filters(
     if tenant_id == PLATFORM_TENANT_FILTER:
         conditions.append(TaskExecution.tenant_id.is_(None))
     elif tenant_id:
-        conditions.append(TaskExecution.tenant_id == tenant_id)  # ty: ignore[invalid-argument-type]
+        conditions.append(col(TaskExecution.tenant_id) == tenant_id)
     if status is not None:
-        conditions.append(TaskExecution.status == status)  # ty: ignore[invalid-argument-type]
+        conditions.append(col(TaskExecution.status) == status)
     if task_name:
         conditions.append(
             TaskExecution.task_name.ilike(like_contains_pattern(task_name), escape=LIKE_ESCAPE_CHAR)
         )
     if queue:
-        conditions.append(TaskExecution.queue == queue)  # ty: ignore[invalid-argument-type]
+        conditions.append(col(TaskExecution.queue) == queue)
     return conditions
 
 

modules/background_tasks/background_tasks/log_purge.py

@@ -6,11 +6,12 @@
 
 from sqlalchemy import delete
 from sqlalchemy.orm import Session
+from sqlmodel import col
 
 from background_tasks.models import RetainedLog
 
 
 def purge_log_records(session: Session, cutoff: datetime) -> int:
     """Delete log lines older than ``cutoff``; returns the count."""
-    result = session.execute(delete(RetainedLog).where(RetainedLog.ts < cutoff))  # ty: ignore[invalid-argument-type]
+    result = session.execute(delete(RetainedLog).where(col(RetainedLog.ts) < cutoff))
     return result.rowcount or 0

modules/background_tasks/background_tasks/log_service.py

@@ -9,6 +9,7 @@
 from simple_module_db import RequestSession
 from simple_module_db.pagination import PageParams, paginate
 from sqlalchemy import false, select
+from sqlmodel import col
 
 from background_tasks.constants import LOG_LEVEL_NAMES
 from background_tasks.contracts.logs import LogItem, LogListResponse
@@ -60,16 +61,18 @@ async def search(
     ) -> LogListResponse:
         where = [*self.scope.conditions()]
         if task_id is not None:
-            where.append(RetainedLog.task_id == task_id)
+            where.append(col(RetainedLog.task_id) == task_id)
         if (level := level_number(min_level)) is not None:
-            where.append(RetainedLog.level >= level)
+            where.append(col(RetainedLog.level) >= level)
         if q:
-            where.append(RetainedLog.message.ilike(_like(q), escape="\\"))
+            where.append(col(RetainedLog.message).ilike(_like(q), escape="\\"))
         if logger_name:
-            where.append(RetainedLog.logger.startswith(logger_name))
+            where.append(col(RetainedLog.logger).startswith(logger_name))
         found = await paginate(
             self.db,
-            select(RetainedLog).where(*where).order_by(RetainedLog.ts.desc(), RetainedLog.id),  # ty: ignore[invalid-argument-type]
+            select(RetainedLog)
+            .where(*where)
+            .order_by(col(RetainedLog.ts).desc(), col(RetainedLog.id)),
             PageParams(page=page, per_page=per_page),
         )
         return LogListResponse(

modules/background_tasks/background_tasks/outbox_claim.py

@@ -20,6 +20,7 @@
 
 from sqlalchemy import or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from sqlmodel import col
 
 from background_tasks.constants import (
     OUTBOX_BACKOFF_BASE_SECONDS,
@@ -103,12 +104,12 @@ async def reclaim_expired(
     async with factory() as session:
         now = now or await db_now(session)
         expired = (
-            OutboxMessage.status == OutboxStatus.RUNNING,
-            or_(OutboxMessage.locked_until < now, OutboxMessage.locked_until.is_(None)),
+            col(OutboxMessage.status) == OutboxStatus.RUNNING,
+            or_(col(OutboxMessage.locked_until) < now, col(OutboxMessage.locked_until).is_(None)),
         )
         dead = await session.execute(
             update(OutboxMessage)
-            .where(*expired, OutboxMessage.attempts >= max_attempts)  # ty: ignore[invalid-argument-type]
+            .where(*expired, col(OutboxMessage.attempts) >= max_attempts)
             .values(
                 status=OutboxStatus.DEAD,
                 locked_until=None,
@@ -118,7 +119,7 @@ async def reclaim_expired(
         )
         back = await session.execute(
             update(OutboxMessage)
-            .where(*expired)  # ty: ignore[invalid-argument-type]
+            .where(*expired)
             .values(status=OutboxStatus.PENDING, locked_until=None, next_attempt_at=now)
             .execution_options(synchronize_session=False)
         )
@@ -174,7 +175,7 @@ async def claim(
             update(OutboxMessage)
             .where(
                 OutboxMessage.id.in_(select(claimable.c.id)),
-                OutboxMessage.status == OutboxStatus.PENDING,  # ty: ignore[invalid-argument-type]
+                col(OutboxMessage.status) == OutboxStatus.PENDING,
             )
             .values(
                 status=OutboxStatus.RUNNING,

modules/background_tasks/background_tasks/outbox_service.py

@@ -9,6 +9,7 @@
 from fastapi import HTTPException, status
 from simple_module_db.pagination import PageParams, paginate
 from sqlalchemy import delete, false, func, select, update
+from sqlmodel import col
 
 from background_tasks.constants import OutboxStatus
 from background_tasks.contracts.outbox import OutboxItem, OutboxListResponse
@@ -66,7 +67,7 @@ async def list(
             self.db,
             select(OutboxMessage)
             .where(*where)
-            .order_by(OutboxMessage.created_at.desc(), OutboxMessage.id),  # ty: ignore[invalid-argument-type]
+            .order_by(col(OutboxMessage.created_at).desc(), col(OutboxMessage.id)),
             PageParams(page=page, per_page=per_page),
         )
         return OutboxListResponse(
@@ -93,7 +94,7 @@ async def _get(self, message_id: uuid.UUID) -> OutboxMessage:
         row = (
             await self.db.execute(
                 select(OutboxMessage)
-                .where(OutboxMessage.id == message_id, *self.scope.conditions())  # ty: ignore[invalid-argument-type]
+                .where(col(OutboxMessage.id) == message_id, *self.scope.conditions())
                 .execution_options(populate_existing=True)
             )
         ).scalar_one_or_none()
@@ -114,8 +115,8 @@ async def retry(self, message_id: uuid.UUID) -> OutboxItem:
         result = await self.db.execute(
             update(OutboxMessage)
             .where(
-                OutboxMessage.id == message_id,  # ty: ignore[invalid-argument-type]
-                OutboxMessage.status == OutboxStatus.DEAD,  # ty: ignore[invalid-argument-type]
+                col(OutboxMessage.id) == message_id,
+                col(OutboxMessage.status) == OutboxStatus.DEAD,
                 *self.scope.conditions(),
             )
             .values(
@@ -141,7 +142,7 @@ async def discard(self, message_id: uuid.UUID) -> None:
         result = await self.db.execute(
             delete(OutboxMessage)
             .where(
-                OutboxMessage.id == message_id,  # ty: ignore[invalid-argument-type]
+                col(OutboxMessage.id) == message_id,
                 OutboxMessage.status.in_(DISCARDABLE),
                 *self.scope.conditions(),
             )

modules/background_tasks/background_tasks/retry_service.py

@@ -16,6 +16,7 @@
 from simple_module_core.events import EventBus
 from simple_module_db import RequestSession, all_tenants, tenant_context
 from sqlalchemy import Select, func, select, update
+from sqlmodel import col
 
 from background_tasks._signal_support import now_utc
 from background_tasks.after_commit import after_commit
@@ -271,8 +272,8 @@ def _fail_unsent(task_id: str, exc: Exception) -> None:
             session.execute(
                 update(TaskExecution)
                 .where(
-                    TaskExecution.celery_task_id == task_id,  # ty: ignore[invalid-argument-type]
-                    TaskExecution.status == TaskStatus.PENDING,  # ty: ignore[invalid-argument-type]
+                    col(TaskExecution.celery_task_id) == task_id,
+                    col(TaskExecution.status) == TaskStatus.PENDING,
                 )
                 .values(
                     status=TaskStatus.FAILED,

modules/background_tasks/background_tasks/task_heartbeat.py

@@ -24,6 +24,7 @@
 
 from celery import current_task
 from sqlalchemy import update
+from sqlmodel import col
 
 from background_tasks._signal_support import now_utc
 from background_tasks.constants import (
@@ -122,11 +123,11 @@ def heartbeat(
             session.execute(
                 update(TaskExecution)
                 .where(
-                    TaskExecution.celery_task_id == task_id,  # ty: ignore[invalid-argument-type]
+                    col(TaskExecution.celery_task_id) == task_id,
                     # Only a row still ``running``: a heartbeat must not revive
                     # a row the sweep already declared stuck (it may have been
                     # retried since) or rewrite a terminal one.
-                    TaskExecution.status == TaskStatus.RUNNING,  # ty: ignore[invalid-argument-type]
+                    col(TaskExecution.status) == TaskStatus.RUNNING,
                 )
                 .values(heartbeat_at=now_utc(), **pending)
             )