@ancient-wolf-19325 can I have a single custom sensor trigger different workflows based on some condition?

It seems likely that you can implement a custom sensor in Flyte that triggers different workflows based on a condition, but this would require writing a user container task plugin (sensor) that checks for the event and then, within the workflow logic, branches to different sub-workflows or tasks depending on the sensor's output. The Flyte platform supports custom sensor plugins and conditional branching in workflows, but there is no explicit built-in mechanism for a sensor to directly trigger multiple workflows; instead, the sensor can be used as a task within a workflow, and the workflow can use conditionals to decide which downstream workflow or task to execute based on the sensor's result. See the user container task plugin example and advanced composition guide. Would you like more detail, including a code example of how to structure such a workflow with a custom sensor and conditional branching? Sources: - user container task plugin example - advanced composition guide 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 Give me code example for both what the sensor would look like, and what the workflows would look like

Certainly! Here’s a concise example: Custom Sensor (User Container Task Plugin): This sensor waits for a file and returns a value that can be used for branching:

from datetime import timedelta
from flytekit import TaskMetadata, task, workflow
from flytekit.extend import Interface, PythonTask, context_manager

class WaitForObjectStoreFile(PythonTask):
    _VAR_NAME: str = "path"

    def __init__(self, name: str, poll_interval: timedelta = timedelta(seconds=10), **kwargs):
        super().__init__(
            task_type="object-store-sensor",
            name=name,
            task_config=None,
            interface=Interface(inputs={self._VAR_NAME: str}, outputs={"result": str}),
            **kwargs,
        )
        self._poll_interval = poll_interval

    def execute(self, **kwargs):
        path = kwargs[self._VAR_NAME]
        ctx = context_manager.FlyteContext.current_context()
        user_context = ctx.user_space_params
        # Simulate sensing logic
        if "trigger_a" in path:
            return {"result": "A"}
        else:
            return {"result": "B"}

sensor = WaitForObjectStoreFile(
    name="my-objectstore-sensor",
    metadata=TaskMetadata(retries=3, timeout=timedelta(minutes=10)),
    poll_interval=timedelta(seconds=1),
)

Workflows with Conditional Branching:

from flytekit import task, workflow

@task
def workflow_a():
    return "Workflow A triggered"

@task
def workflow_b():
    return "Workflow B triggered"

@workflow
def main_wf(path: str) -> str:
    sensor_result = sensor(path=path)
    if sensor_result["result"] == "A":
        return workflow_a()
    else:
        return workflow_b()

This setup allows the sensor to determine which downstream workflow to trigger based on its output. You can run

main_wf(path="trigger_a_file")

or

main_wf(path="other_file")

to see different branches. Sources: - user container task plugin example - Flytesnacks user container example 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 so sensor is connected a main workflow, which then triggers other workflows based on the condition?

My understanding is that, yes, the sensor is typically used as a task within a main workflow, and then the main workflow uses the sensor’s output to conditionally trigger other workflows or tasks. The sensor itself does not directly trigger workflows; instead, it provides information (such as a result or status) that the main workflow uses in its branching logic to decide which downstream workflow or task to execute. This pattern is supported by Flyte’s extensible task and workflow model, as shown in the user container task plugin example and advanced composition guide. Would you like more detail or a more advanced example? Sources: - user container task plugin example 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@freezing-airport-6809 (sry tagging you because I don't know who else to tag) I'm not sure if runllm is hallucinating here. Is what I'm trying to do possible?

i dont understand your question