style(triggers): clear ruff findings on new modules

Cosmetic-only changes from running ruff --fix and a handful of manual
fixups:

  - D205: split multi-line docstring openers into single-line
    summaries followed by a blank line and the longer description.
  - SIM105: use contextlib.suppress(asyncio.TimeoutError) in
    _sleep_with_stop instead of try/except/pass.
  - RUF015: avoid list(...)[0] in the trigger_job assertion, bind
    once and index.
  - TC003: move uuid.UUID import in crud.py into the TYPE_CHECKING
    block since it is only referenced as a parameter annotation.

No behaviour changes. All 15 unit tests still pass and ruff is now
clean on every file touched by this feature branch.

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

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
 
 

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

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
 
 
 # --------------------------------------------------------------------------- #

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

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)