From e136bc7e709cd0f4e5fa853d5948ea7fb9d32242 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Thu, 21 May 2026 15:09:04 -0300 Subject: [PATCH] chore(triggers): silence SQLModel exec-warning on legitimate execute() calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../services/database/models/triggers/crud.py | 7 ++- .../services/triggers/_sqlmodel_compat.py | 44 +++++++++++++++++++ .../base/langflow/services/triggers/worker.py | 22 ++++++---- 3 files changed, 62 insertions(+), 11 deletions(-) create mode 100644 src/backend/base/langflow/services/triggers/_sqlmodel_compat.py diff --git a/src/backend/base/langflow/services/database/models/triggers/crud.py b/src/backend/base/langflow/services/database/models/triggers/crud.py index 96eb57ed13..81bcb0decf 100644 --- a/src/backend/base/langflow/services/database/models/triggers/crud.py +++ b/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 diff --git a/src/backend/base/langflow/services/triggers/_sqlmodel_compat.py b/src/backend/base/langflow/services/triggers/_sqlmodel_compat.py new file mode 100644 index 0000000000..1639b81356 --- /dev/null +++ b/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 diff --git a/src/backend/base/langflow/services/triggers/worker.py b/src/backend/base/langflow/services/triggers/worker.py index 29e0cb90a0..433cd164f0 100644 --- a/src/backend/base/langflow/services/triggers/worker.py +++ b/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