From 18cd0c1aa572db40ea05d7a61629c9b4fdbc4eb0 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Thu, 21 May 2026 13:57:34 -0300 Subject: [PATCH] feat(triggers): add CronTrigger component and triggers category MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The trigger is now an in-flow component, not an external entity. Dragging CronTrigger into a flow declares the schedule; removing it removes the schedule. The component's class name ("CronTrigger") is the immutable identifier persisted in node ids — pinned by a unit test so any accidental rename surfaces as a test failure rather than breaking every saved flow. Files (all new): src/lfx/src/lfx/components/triggers/cron_trigger.py The Component class. Five inputs (cron_expression, timezone dropdown with combobox=True for free-text IANA, max_attempts with RangeSpec(1..10), payload JSON, worker-fed fire_time) and a single Message output. build_event() echoes the worker-injected fire_time when present and falls back to the current UTC instant for manual canvas runs — mirroring the Webhook component's 'inert outside the trigger path' contract. src/lfx/src/lfx/components/triggers/constants.py DRY landing pad for the curated IANA list, default cron, default timezone, default max_attempts, and the upper attempt limit. The worker and any future trigger consumer import the same constants, so a change here propagates everywhere. src/lfx/src/lfx/components/triggers/__init__.py Lazy import pattern identical to every other category. Three single-line additions to src/lfx/src/lfx/components/__init__.py register the new category in (1) _dynamic_imports, (2) __all__, and (3) the TYPE_CHECKING block. No other existing file touched. Tests (new): src/backend/tests/unit/components/test_cron_trigger.py covers class identity, palette metadata, input names + defaults, timezone dropdown contents, output shape, manual run path, worker injection path, status message, and whitespace defensiveness. 10/10 pass; ruff clean. --- .../unit/components/test_cron_trigger.py | 111 +++++++++++++++ src/lfx/src/lfx/components/__init__.py | 3 + .../src/lfx/components/triggers/__init__.py | 43 ++++++ .../src/lfx/components/triggers/constants.py | 40 ++++++ .../lfx/components/triggers/cron_trigger.py | 132 ++++++++++++++++++ 5 files changed, 329 insertions(+) create mode 100644 src/backend/tests/unit/components/test_cron_trigger.py create mode 100644 src/lfx/src/lfx/components/triggers/__init__.py create mode 100644 src/lfx/src/lfx/components/triggers/constants.py create mode 100644 src/lfx/src/lfx/components/triggers/cron_trigger.py diff --git a/src/backend/tests/unit/components/test_cron_trigger.py b/src/backend/tests/unit/components/test_cron_trigger.py new file mode 100644 index 0000000000..3964dbf05e --- /dev/null +++ b/src/backend/tests/unit/components/test_cron_trigger.py @@ -0,0 +1,111 @@ +"""Unit tests for the CronTrigger component. + +Pure instantiation tests — no graph wiring, no DB. Verifies the +component declares the immutable identifier, the expected inputs and +output, and that the two execution paths (manual canvas run vs +worker-injected fire) both produce a tz-aware ISO 8601 string message. +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from lfx.components.triggers import CronTriggerComponent +from lfx.components.triggers.constants import ( + COMMON_TIMEZONES, + DEFAULT_CRON_EXPRESSION, + DEFAULT_MAX_ATTEMPTS, + DEFAULT_TIMEZONE, +) +from lfx.schema.message import Message + + +def test_class_identity_is_immutable(): + # ``name`` is persisted in the node id of every saved flow. + # Pinning the value here makes any accidental rename a test failure. + assert CronTriggerComponent.name == "CronTrigger" + + +def test_metadata_for_palette(): + assert CronTriggerComponent.display_name == "Cron Trigger" + assert CronTriggerComponent.icon == "clock" + assert "cron" in CronTriggerComponent.description.lower() + + +def test_inputs_present_with_expected_names(): + component = CronTriggerComponent() + names = {i.name for i in component.inputs} + assert names == {"cron_expression", "timezone", "max_attempts", "payload", "fire_time"} + + +def test_inputs_carry_sensible_defaults(): + component = CronTriggerComponent() + by_name = {i.name: i for i in component.inputs} + assert by_name["cron_expression"].value == DEFAULT_CRON_EXPRESSION + assert by_name["timezone"].value == DEFAULT_TIMEZONE + assert by_name["max_attempts"].value == DEFAULT_MAX_ATTEMPTS + + +def test_timezone_dropdown_lists_common_iana_names(): + component = CronTriggerComponent() + by_name = {i.name: i for i in component.inputs} + options = by_name["timezone"].options + # Combobox is true so other zones are allowed; the curated list + # must at least include UTC and a few well-known regions. + assert "UTC" in options + assert "America/Sao_Paulo" in options + assert "Europe/London" in options + # The exported tuple should match the dropdown one-to-one. + assert set(options) == set(COMMON_TIMEZONES) + + +def test_output_is_single_message_emitter(): + component = CronTriggerComponent() + assert len(component.outputs) == 1 + only = component.outputs[0] + assert only.name == "event" + assert only.method == "build_event" + + +def test_manual_canvas_run_returns_current_utc(): + """No fire_time → component emits the call instant. + + The exact value is unstable across runs, so we assert it parses + as ISO 8601 and is recent (within 5 seconds of the test's clock). + """ + component = CronTriggerComponent() + before = datetime.now(timezone.utc) + message = component.build_event() + after = datetime.now(timezone.utc) + assert isinstance(message, Message) + parsed = datetime.fromisoformat(message.text) + assert before <= parsed <= after + assert parsed.tzinfo is not None + + +def test_worker_injection_path_emits_provided_timestamp(): + """Worker-set fire_time → component echoes it verbatim. + + This is the contract relied on by the worker dispatcher: whatever + tz-aware ISO string the worker writes into the tweak shows up in + the downstream Message unchanged. + """ + component = CronTriggerComponent() + component.fire_time = "2026-05-21T12:34:56+00:00" + message = component.build_event() + assert message.text == "2026-05-21T12:34:56+00:00" + + +def test_status_string_includes_fire_time(): + component = CronTriggerComponent() + component.fire_time = "2026-05-21T12:34:56+00:00" + component.build_event() + assert "2026-05-21T12:34:56+00:00" in str(component.status) + + +def test_fire_time_whitespace_is_ignored(): + """Defensive: a worker that injects " ... " still produces a clean message.""" + component = CronTriggerComponent() + component.fire_time = " 2026-05-21T12:34:56+00:00 " + message = component.build_event() + assert message.text == "2026-05-21T12:34:56+00:00" diff --git a/src/lfx/src/lfx/components/__init__.py b/src/lfx/src/lfx/components/__init__.py index 148245bae5..a04b99b120 100644 --- a/src/lfx/src/lfx/components/__init__.py +++ b/src/lfx/src/lfx/components/__init__.py @@ -92,6 +92,7 @@ if TYPE_CHECKING: textsplitters, toolkits, tools, + triggers, twelvelabs, unstructured, upstash, @@ -197,6 +198,7 @@ _dynamic_imports = { "textsplitters": "__module__", "toolkits": "__module__", "tools": "__module__", + "triggers": "__module__", "twelvelabs": "__module__", "unstructured": "__module__", "upstash": "__module__", @@ -328,6 +330,7 @@ __all__ = [ "textsplitters", "toolkits", "tools", + "triggers", "twelvelabs", "unstructured", "upstash", diff --git a/src/lfx/src/lfx/components/triggers/__init__.py b/src/lfx/src/lfx/components/triggers/__init__.py new file mode 100644 index 0000000000..9442e77377 --- /dev/null +++ b/src/lfx/src/lfx/components/triggers/__init__.py @@ -0,0 +1,43 @@ +"""Trigger components. + +A trigger component fires the containing flow on some external signal +(time, queue, event). The ``CronTrigger`` is the time-based one. + +Lazy import pattern matches every other category: ``_dynamic_imports`` +maps the class name to its module file and the global ``__getattr__`` +in ``lfx.components.__init__`` does the actual import on first +attribute access. This keeps server cold-start times low when most +trigger types are not used. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from lfx.components._importing import import_mod + +if TYPE_CHECKING: + from .cron_trigger import CronTriggerComponent + +_dynamic_imports: dict[str, str] = { + "CronTriggerComponent": "cron_trigger", +} + +__all__ = ["CronTriggerComponent"] + + +def __getattr__(attr_name: str) -> Any: + if attr_name not in _dynamic_imports: + msg = f"module '{__name__}' has no attribute '{attr_name}'" + raise AttributeError(msg) + try: + result = import_mod(attr_name, _dynamic_imports[attr_name], __spec__.parent) + except (ModuleNotFoundError, ImportError, AttributeError) as e: + msg = f"Could not import '{attr_name}' from '{__name__}': {e}" + raise AttributeError(msg) from e + globals()[attr_name] = result + return result + + +def __dir__() -> list[str]: + return list(__all__) diff --git a/src/lfx/src/lfx/components/triggers/constants.py b/src/lfx/src/lfx/components/triggers/constants.py new file mode 100644 index 0000000000..0336541345 --- /dev/null +++ b/src/lfx/src/lfx/components/triggers/constants.py @@ -0,0 +1,40 @@ +"""Constants for the triggers component category. + +Kept in its own module so the component file stays focused on the +``Component`` declaration and any downstream consumer (worker, +discovery helper) can import the same authoritative list. +""" + +from __future__ import annotations + +# Curated subset of IANA timezone names exposed in the CronTrigger +# dropdown. The list is intentionally short — a sensible default that +# covers the most common regions. Users needing a timezone outside +# this set can type any IANA name via the combobox mode. +# +# The order is roughly geographic (Americas → Europe → Asia → +# Australia → UTC last) so the dropdown reads naturally; UTC is also +# the dropdown default and the safe fallback elsewhere in the system. +COMMON_TIMEZONES: tuple[str, ...] = ( + "UTC", + "America/New_York", + "America/Chicago", + "America/Denver", + "America/Los_Angeles", + "America/Sao_Paulo", + "America/Argentina/Buenos_Aires", + "America/Mexico_City", + "Europe/London", + "Europe/Berlin", + "Europe/Paris", + "Europe/Madrid", + "Asia/Tokyo", + "Asia/Shanghai", + "Asia/Kolkata", + "Australia/Sydney", +) + +DEFAULT_TIMEZONE: str = "UTC" +DEFAULT_CRON_EXPRESSION: str = "*/5 * * * *" +DEFAULT_MAX_ATTEMPTS: int = 3 +MAX_ATTEMPTS_LIMIT: int = 10 diff --git a/src/lfx/src/lfx/components/triggers/cron_trigger.py b/src/lfx/src/lfx/components/triggers/cron_trigger.py new file mode 100644 index 0000000000..d73a107fb1 --- /dev/null +++ b/src/lfx/src/lfx/components/triggers/cron_trigger.py @@ -0,0 +1,132 @@ +"""Cron Trigger component. + +A trigger that fires the surrounding flow on a recurring cron schedule. +The component lives in the flow canvas like any other node; its +configuration (``cron_expression``, ``timezone``) is the source of +truth and is read by the in-process trigger worker — there is no +parallel registry table. + +Design notes: + +* ``name = "CronTrigger"`` is the immutable identifier. It is used + both by the backend detection helper (``services.triggers.discovery``) + and as the prefix of the node id in ``flow.data`` (e.g. + ``"CronTrigger-abc12"``). Renaming would be a breaking change for + every saved flow. +* The ``fire_time`` input is invisible to manual canvas runs. The + worker injects it via the same tweak mechanism the webhook handler + uses, so the component can pass the actual fire instant downstream + without needing a dedicated runtime channel. +* Manual canvas runs (Play button) are a no-op: the component emits + the current UTC time as the event message. That's deliberate — + matches the Webhook behaviour, which is also inert outside its HTTP + trigger path. +* No ``is_active`` field: the presence of the node in the flow IS the + activation. To pause a trigger, the user removes the node (or the + containing flow). Simpler invariant, less to keep in sync. +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from lfx.custom.custom_component.component import Component +from lfx.field_typing.range_spec import RangeSpec +from lfx.io import ( + DropdownInput, + IntInput, + MessageTextInput, + MultilineInput, + Output, +) +from lfx.schema.message import Message + +from .constants import ( + COMMON_TIMEZONES, + DEFAULT_CRON_EXPRESSION, + DEFAULT_MAX_ATTEMPTS, + DEFAULT_TIMEZONE, + MAX_ATTEMPTS_LIMIT, +) + + +class CronTriggerComponent(Component): + display_name = "Cron Trigger" + description = "Fire this flow on a recurring cron schedule." + documentation: str = "https://docs.langflow.org/component-cron-trigger" + # The class name pinned by the persisted node id. Do not rename. + name = "CronTrigger" + icon = "clock" + + inputs = [ + MessageTextInput( + name="cron_expression", + display_name="Cron Expression", + info=( + "Standard 5-field POSIX cron (minute hour day month weekday). " + "Example: '*/5 * * * *' fires every five minutes." + ), + value=DEFAULT_CRON_EXPRESSION, + required=True, + # No upstream connection — this is a config value, not data. + input_types=[], + ), + DropdownInput( + name="timezone", + display_name="Timezone", + info=( + "IANA timezone name used to interpret the cron expression. " + "Type any IANA name to use a timezone not in the list." + ), + options=list(COMMON_TIMEZONES), + value=DEFAULT_TIMEZONE, + combobox=True, + ), + IntInput( + name="max_attempts", + display_name="Max Attempts", + info="Retry budget per fire. Failed runs are retried with exponential backoff up to this many times.", + value=DEFAULT_MAX_ATTEMPTS, + range_spec=RangeSpec(min=1, max=MAX_ATTEMPTS_LIMIT, step=1), + advanced=True, + ), + MultilineInput( + name="payload", + display_name="Payload (JSON)", + info=( + "Optional JSON object merged into the SimplifiedAPIRequest fields " + "(input_value, input_type, output_type, tweaks, session_id) when the trigger fires." + ), + advanced=True, + input_types=[], + ), + # Worker-populated. Empty on manual canvas runs. + MessageTextInput( + name="fire_time", + display_name="Fire Time", + info="Set by the trigger worker at fire time. Empty on manual runs.", + advanced=True, + input_types=[], + ), + ] + + outputs = [ + Output( + display_name="Trigger Event", + name="event", + method="build_event", + ), + ] + + def build_event(self) -> Message: + """Emit the event message that downstream nodes consume. + + Returns the worker-injected ``fire_time`` when present, otherwise + the current UTC instant. Either value is a tz-aware ISO 8601 + string, so downstream components see a consistent shape + regardless of how the flow was kicked off. + """ + fire_time = (self.fire_time or "").strip() + text = fire_time or datetime.now(timezone.utc).isoformat() + self.status = f"Cron trigger fired at {text}" + return Message(text=text)