diff --git a/src/backend/base/langflow/alembic/versions/tg01a2b3c4d5_add_native_triggers_schema.py b/src/backend/base/langflow/alembic/versions/tg01a2b3c4d5_add_native_triggers_schema.py index 0f85ec38ca..9fbf7a8784 100644 --- a/src/backend/base/langflow/alembic/versions/tg01a2b3c4d5_add_native_triggers_schema.py +++ b/src/backend/base/langflow/alembic/versions/tg01a2b3c4d5_add_native_triggers_schema.py @@ -21,9 +21,8 @@ from collections.abc import Sequence import sqlalchemy as sa from alembic import op -from sqlalchemy.dialects.postgresql import JSONB - from langflow.utils import migration +from sqlalchemy.dialects.postgresql import JSONB # revision identifiers, used by Alembic. revision: str = "tg01a2b3c4d5" # pragma: allowlist secret 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 d0e93f0c31..5425e18bb7 100644 --- a/src/backend/base/langflow/services/database/models/triggers/crud.py +++ b/src/backend/base/langflow/services/database/models/triggers/crud.py @@ -9,7 +9,6 @@ from __future__ import annotations from datetime import datetime, timezone from typing import TYPE_CHECKING -from uuid import UUID from sqlalchemy import update from sqlmodel import col, select @@ -18,6 +17,8 @@ from langflow.services.database.models.jobs.model import JobStatus from langflow.services.database.models.triggers.model import Trigger, TriggerJob if TYPE_CHECKING: + from uuid import UUID + from sqlmodel.ext.asyncio.session import AsyncSession diff --git a/src/backend/base/langflow/services/database/models/triggers/model.py b/src/backend/base/langflow/services/database/models/triggers/model.py index 25dcf99a80..c59d4ef1d8 100644 --- a/src/backend/base/langflow/services/database/models/triggers/model.py +++ b/src/backend/base/langflow/services/database/models/triggers/model.py @@ -251,8 +251,10 @@ class TriggerUpdate(SQLModel): class TriggerRead(SQLModel): - """Response shape for trigger reads. Excludes nothing — all columns are - safe to surface (no secrets stored on this table).""" + """Response shape for trigger reads. + + All columns are safe to surface — no secrets are stored on this table. + """ id: UUID flow_id: UUID diff --git a/src/backend/base/langflow/services/triggers/worker.py b/src/backend/base/langflow/services/triggers/worker.py index 5006d48db6..f9b979570c 100644 --- a/src/backend/base/langflow/services/triggers/worker.py +++ b/src/backend/base/langflow/services/triggers/worker.py @@ -32,6 +32,7 @@ single bad trigger never takes the worker down. from __future__ import annotations import asyncio +import contextlib from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING @@ -82,9 +83,9 @@ def _retry_delay(attempt: int) -> timedelta: async def _claim_one(session: AsyncSession) -> _ClaimedJob | None: - """Atomically pick the next eligible ``trigger_job`` row and flip it - to ``in_progress``. + """Atomically claim the next eligible ``trigger_job`` row. + Flips the chosen row to ``in_progress``. On Postgres, uses ``FOR UPDATE SKIP LOCKED`` so multiple workers can run concurrently without ever handing the same row to two workers. On SQLite, uses an optimistic ``UPDATE ... WHERE status = @@ -174,8 +175,10 @@ async def _load_trigger_and_flow( session: AsyncSession, trigger_id: UUID, ) -> tuple[Trigger, Flow, User] | None: - """Return the trigger + its flow + its owning user, or ``None`` if any - of the three was deleted between claim and dispatch.""" + """Return the trigger plus its flow and owning user. + + Returns ``None`` if any of the three was deleted between claim and dispatch. + """ trigger = (await session.exec(select(Trigger).where(Trigger.id == trigger_id))).first() if trigger is None: return None @@ -393,10 +396,8 @@ async def trigger_worker_loop(stop_event: asyncio.Event) -> None: async def _sleep_with_stop(stop_event: asyncio.Event, seconds: float) -> None: """Sleep but wake immediately if the stop event is set.""" - try: + with contextlib.suppress(asyncio.TimeoutError): await asyncio.wait_for(stop_event.wait(), timeout=seconds) - except asyncio.TimeoutError: # noqa: PERF203 — TimeoutError is the happy path - pass # --------------------------------------------------------------------------- # diff --git a/src/backend/tests/unit/services/triggers/test_scheduler.py b/src/backend/tests/unit/services/triggers/test_scheduler.py index 5dc4ddbeac..debc458dad 100644 --- a/src/backend/tests/unit/services/triggers/test_scheduler.py +++ b/src/backend/tests/unit/services/triggers/test_scheduler.py @@ -11,7 +11,6 @@ from __future__ import annotations from datetime import datetime, timezone import pytest - from langflow.services.triggers.scheduler import ( InvalidTriggerConfigError, next_fire_time_utc, @@ -64,8 +63,10 @@ def test_dst_spring_forward_skips_phantom_local_time(): def test_cron_rolls_into_next_day(): - """09:00 Europe/London cron with reference at noon UTC should land - on the next day's 09:00 London time (=08:00 UTC in BST).""" + """09:00 Europe/London cron with reference at noon UTC rolls forward. + + Lands on the next day's 09:00 London time (=08:00 UTC in BST). + """ after = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) result = next_fire_time_utc( cron_expression="0 9 * * *", @@ -99,8 +100,10 @@ def test_validate_trigger_config_bundles_both_checks(): def test_next_fire_time_with_default_after_returns_future(): - """When ``after`` is omitted the function uses ``utcnow``; the - returned datetime must be strictly later than the call instant.""" + """Omitting ``after`` should default to ``utcnow``. + + The returned datetime must be strictly later than the call instant. + """ before_call = datetime.now(timezone.utc) result = next_fire_time_utc(cron_expression="* * * * *", timezone_name="UTC") assert result > before_call diff --git a/src/backend/tests/unit/services/triggers/test_worker.py b/src/backend/tests/unit/services/triggers/test_worker.py index 7c3fa13e5e..d4546d7d6c 100644 --- a/src/backend/tests/unit/services/triggers/test_worker.py +++ b/src/backend/tests/unit/services/triggers/test_worker.py @@ -13,13 +13,12 @@ from datetime import datetime, timedelta, timezone from uuid import uuid4 import pytest -from sqlmodel import select - from langflow.services.database.models.flow.model import Flow from langflow.services.database.models.jobs.model import JobStatus from langflow.services.database.models.triggers import Trigger, TriggerJob, TriggerType from langflow.services.database.models.user.model import User from langflow.services.triggers import worker as worker_module +from sqlmodel import select def _utcnow() -> datetime: @@ -154,8 +153,9 @@ async def test_finalize_success_marks_completed_and_enqueues_next_for_recurring( ), ) ).all() - assert len(list(queued)) == 1 - assert list(queued)[0].id != job.id + queued_list = list(queued) + assert len(queued_list) == 1 + assert queued_list[0].id != job.id @pytest.mark.asyncio @@ -218,9 +218,11 @@ async def test_finalize_failure_enqueues_retry_when_budget_remains(async_session @pytest.mark.asyncio async def test_finalize_failure_at_max_attempts_enqueues_only_next_cron(async_session): - """When the retry budget is exhausted, the failed row stays FAILED - and the only queued follow-up is the next cron fire — not another - retry of the same attempt chain.""" + """When the retry budget is exhausted, the failed row stays FAILED. + + The only queued follow-up is the next cron fire — never another retry + of the same attempt chain. + """ user, flow = await _seed_user_and_flow(async_session) trigger = await _make_trigger(async_session, user=user, flow=flow, max_attempts=2) job = await _enqueue_job(async_session, trigger=trigger, attempt=2)