chore(triggers): silence SQLModel exec-warning on legitimate execute() calls

SQLModel's AsyncSession raises a DeprecationWarning on every
session.execute() call urging the caller to use session.exec() —
but exec is typed for SELECT-returning statements only. Our five
execute() call-sites all pass non-SELECT statements that exec
explicitly does not accept (text() raw SQL or update() constructs),
so the warning is informational and cannot be acted on by changing
the code.

In the worker's hot path the warning fires on every claim cycle
(every 0.5–5s depending on backoff), flooding the log. The crud
helpers fire it once at boot (reset_stalled_in_progress) and once
per flow save (cancel_queued_jobs_for_components). None are
actionable, all are noisy.

NEW services/triggers/_sqlmodel_compat.py
  suppress_sqlmodel_exec_warning() — a narrow context manager that
  filters DeprecationWarning ONLY from sqlmodel.ext.asyncio.session.
  Any other DeprecationWarning still surfaces. The module is
  underscore-prefixed to signal it's an internal compatibility shim,
  not part of the public surface.

Five call-sites wrapped:
  worker.py (3): _claim_one's Postgres FOR UPDATE SKIP LOCKED select,
                its follow-up UPDATE, and the SQLite optimistic
                UPDATE...RETURNING.
  crud.py (2):  cancel_queued_jobs_for_components and
                reset_stalled_in_progress, both update() constructs.

Verified by running the full trigger test suite with
-W 'error::DeprecationWarning:sqlmodel.ext.asyncio.session' — all
tests still pass, confirming the suppression actually engages at
runtime (without it, the warnings-as-errors filter would fail every
test that hits the worker path). Ruff clean on all three files.

src/backend/base/langflow/services/database/models/triggers/crud.py

@@ -16,6 +16,7 @@ from sqlmodel import col, select
 
 from langflow.services.database.models.jobs.model import JobStatus
 from langflow.services.database.models.triggers.model import TriggerJob
+from langflow.services.triggers._sqlmodel_compat import suppress_sqlmodel_exec_warning
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -70,7 +71,8 @@ async def cancel_queued_jobs_for_components(
         )
         .values(status=JobStatus.CANCELLED, finished_at=now)
     )
-    result = await session.execute(statement)
+    with suppress_sqlmodel_exec_warning():
+        result = await session.execute(statement)
     return result.rowcount or 0
 
 
@@ -95,5 +97,6 @@ async def reset_stalled_in_progress(
         )
         .values(status=JobStatus.QUEUED, scheduled_at=now, started_at=None)
     )
-    result = await session.execute(statement)
+    with suppress_sqlmodel_exec_warning():
+        result = await session.execute(statement)
     return result.rowcount or 0

src/backend/base/langflow/services/triggers/_sqlmodel_compat.py

@@ -0,0 +1,44 @@
+"""Compatibility helpers around SQLModel quirks.
+
+SQLModel's ``AsyncSession.execute`` is the right choice — and the only
+choice — for ``text()`` raw SQL and for ``update()`` / ``delete()``
+constructs. The library still raises a ``DeprecationWarning`` on
+every ``execute`` call urging the caller to use ``exec`` instead, even
+when ``exec`` does not accept the statement type. The warning is
+informational and unrelated to anything actionable on our end.
+
+This module centralises a tight context manager that silences only
+that specific warning, so every trigger module can wrap its raw-SQL
+``execute`` calls without each one re-implementing the suppression.
+
+The filter is narrow on purpose: scoped to the SQLModel module path
+so any other ``DeprecationWarning`` from somewhere else still
+surfaces.
+"""
+
+from __future__ import annotations
+
+import warnings
+from contextlib import contextmanager
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+@contextmanager
+def suppress_sqlmodel_exec_warning() -> Iterator[None]:
+    """Silence SQLModel's 'use exec()' nudge for the wrapped block.
+
+    Use around the few legitimate ``session.execute(...)`` calls we
+    have (raw SQL via :func:`sqlalchemy.text`, or ORM
+    ``update()`` / ``delete()`` constructs that :meth:`AsyncSession.exec`
+    explicitly does not accept).
+    """
+    with warnings.catch_warnings():
+        warnings.filterwarnings(
+            "ignore",
+            category=DeprecationWarning,
+            module=r"sqlmodel\.ext\.asyncio\.session",
+        )
+        yield

src/backend/base/langflow/services/triggers/worker.py

@@ -51,6 +51,7 @@ from langflow.services.database.models.jobs.model import JobStatus
 from langflow.services.database.models.triggers import TriggerJob
 from langflow.services.database.models.user.model import User
 from langflow.services.deps import session_scope
+from langflow.services.triggers._sqlmodel_compat import suppress_sqlmodel_exec_warning
 from langflow.services.triggers.discovery import (
     CronTriggerConfig,
     find_cron_trigger_configs,
@@ -131,17 +132,19 @@ async def _claim_one(session: AsyncSession) -> _ClaimedJob | None:
             "LIMIT 1 "
             "FOR UPDATE SKIP LOCKED"
         )
-        row = (await session.execute(select_stmt, {"now": now})).first()
+        with suppress_sqlmodel_exec_warning():
+            row = (await session.execute(select_stmt, {"now": now})).first()
         if row is None:
             return None
         trigger_job_id, flow_id, component_id, attempt, max_attempts = row
-        await session.execute(
-            text(
-                "UPDATE trigger_job SET status = 'in_progress', started_at = :now "
-                "WHERE id = :id"
-            ),
-            {"now": now, "id": trigger_job_id},
-        )
+        with suppress_sqlmodel_exec_warning():
+            await session.execute(
+                text(
+                    "UPDATE trigger_job SET status = 'in_progress', started_at = :now "
+                    "WHERE id = :id"
+                ),
+                {"now": now, "id": trigger_job_id},
+            )
         return _ClaimedJob(
             trigger_job_id=_coerce_uuid(trigger_job_id),
             flow_id=_coerce_uuid(flow_id),
@@ -163,7 +166,8 @@ async def _claim_one(session: AsyncSession) -> _ClaimedJob | None:
         "AND status = 'queued' "
         "RETURNING id, flow_id, component_id, attempt, max_attempts"
     )
-    result = await session.execute(update_stmt, {"now": now})
+    with suppress_sqlmodel_exec_warning():
+        result = await session.execute(update_stmt, {"now": now})
     row = result.first()
     if row is None:
         return None