[namespaced events] (#2742)

- Make @trigger and ArgoEvent support namespaced eventing. This means
users can do the following:

Have an upstream frow raise an ArgoEvent which is caught by a downstream
flow under the same project and branch. For example, below flow raises
the event.

```python
# python pub.py --branch foo run 
from metaflow import FlowSpec, step, project, namespaced_event_name
from metaflow.integrations import ArgoEvent

@project(name="myproject")
class MyFlow(FlowSpec):
    @step
    def start(self):
        # Publish a namespaced event: mfns.myproject.test.foo.data_ready
        ArgoEvent(namespaced_event_name("data_ready")).publish()
        self.next(self.end)

    @step
    def end(self):
        pass
```

And this one catches that event:
```python
# python pub.py --branch foo argo-workflows create 
@project(name="myproject")
@trigger(event=namespaced_event_name("data_ready"))
class DownstreamFlow(FlowSpec):
    ...
```
- Each event name returned by `namespaced_event_name` is blindly
prefixed with `mfns.` to distinguish namespaced events (this is
configurable via `METAFLOW_NAMESPACED_EVENTS_PREFIX`). this acts as a
proxy for knowing that "metaflow" raised the event.
- The intention of this change is to allow users to enforce
catching/raising events only within the same project/branch namespace to
make experimentation much easier.
- We have a [custom decorator
implementation](https://github.com/outerbounds/custom-decorator-examples/tree/main/namespaced_events)
of this but it is not as natural as a core change.
- We rely on the same machinery introduced in #2157
which allowed @trigger event/s parameters to be DeployTimeFields. We
just lift this information / functionality into the docstring too and
make it a little more public.

Author: valayDave

metaflow/metaflow_config.py

@@ -455,6 +455,9 @@
     "ARGO_EVENTS_SENSOR_NAMESPACE", KUBERNETES_NAMESPACE
 )
 
+# Prefix for namespaced events (used by @trigger with namespaced=True)
+NAMESPACED_EVENTS_PREFIX = from_conf("NAMESPACED_EVENTS_PREFIX", "mfns")
+
 ARGO_WORKFLOWS_UI_URL = from_conf("ARGO_WORKFLOWS_UI_URL")
 
 ##

metaflow/plugins/__init__.py

@@ -174,6 +174,7 @@
 TL_PLUGINS_DESC = [
     ("yaml_parser", ".parsers.yaml_parser"),
     ("requirements_txt_parser", ".pypi.parsers.requirements_txt_parser"),
+    ("namespaced_event_name", ".namespaced_events.namespaced_event_name"),
     ("pyproject_toml_parser", ".pypi.parsers.pyproject_toml_parser"),
     ("conda_environment_yml_parser", ".pypi.parsers.conda_environment_yml_parser"),
 ]

metaflow/plugins/argo/argo_events.py

@@ -27,8 +27,8 @@ class ArgoEvent(object):
 
     Parameters
     ----------
-    name : str,
-        Name of the event
+    name : Union[str, Callable[[], str]]
+        Name of the event, or a callable (invoked with no arguments) that returns the event name (e.g., `namespaced_event_name('foo')`).
     url : str, optional
         Override the event endpoint from `ARGO_EVENTS_WEBHOOK_URL`.
     payload : Dict, optional
@@ -39,6 +39,13 @@ def __init__(
         self, name, url=ARGO_EVENTS_WEBHOOK_URL, payload=None, access_token=None
     ):
         # TODO: Introduce support for NATS
+        if callable(name):
+            name = name()
+            if not isinstance(name, str):
+                raise ArgoEventException(
+                    "Callable for 'name' must return a string, got %s"
+                    % type(name).__name__
+                )
         self._name = name
         self._url = url
         self._payload = payload or {}

metaflow/plugins/events_decorator.py

@@ -5,7 +5,7 @@
 from metaflow.decorators import FlowDecorator
 from metaflow.exception import MetaflowException
 from metaflow.util import is_stringish
-from metaflow.parameters import DeployTimeField, deploy_time_eval
+from metaflow.parameters import DeployTimeField, deploy_time_eval, ParameterContext
 
 # TODO: Support dynamic parameter mapping through a context object that exposes
 #       flow name and user name similar to parameter context
@@ -43,12 +43,23 @@ class TriggerDecorator(FlowDecorator):
     @trigger(event={'name':'foo', 'parameters':{'common_name': 'common_name', 'flow_param': 'event_field'}})
     ```
 
+    For namespaced events, you can use `namespaced_event_name` which resolves the
+    full event name at deploy time based on @project settings:
+    ```
+    from metaflow import namespaced_event_name
+
+    @trigger(event=namespaced_event_name('foo'))
+
+    @trigger(event={'name': namespaced_event_name('foo'), 'parameters': {'x': 'y'}})
+    ```
+
     Parameters
     ----------
-    event : Union[str, Dict[str, Any]], optional, default None
-        Event dependency for this flow.
-    events : List[Union[str, Dict[str, Any]]], default []
-        Events dependency for this flow.
+    event : Union[str, Dict[str, Any], Callable[[ParameterContext], Union[str, Dict[str, Any]]]], optional, default None
+        Event dependency for this flow. Can be a string, dict, or a callable that
+        returns a string or dict at deploy time.
+    events : List[Union[str, Dict[str, Any],Callable[[ParameterContext], Union[str, Dict[str, Any]]]]], default []
+        Events dependency for this flow. Each element can be a string, dict, or callable.
     options : Dict[str, Any], default {}
         Backend-specific configuration for tuning eventing behavior.
 

metaflow/plugins/namespaced_events.py

@@ -0,0 +1,105 @@
+from collections import namedtuple
+import functools
+import re
+
+from metaflow.metaflow_config import NAMESPACED_EVENTS_PREFIX
+from metaflow.util import is_stringish
+
+ParsedEvent = namedtuple(
+    "ParsedEvent",
+    ["full_name", "project", "branch", "logical_name", "is_namespaced"],
+)
+
+
+def namespaced_event_name(event_name):
+    """
+    Creates a project-namespaced event name based on @project settings.
+
+    Use this to automatically prefix event names with your @project
+    namespace, ensuring events are isolated per project and branch.
+
+    The resolved name follows the format:
+        mfns.<project>.<branch>.<event_name>
+
+    If no @project decorator is present, resolves to:
+        mfns.<event_name>
+
+    Parameters
+    ----------
+    event_name : str
+        The logical event name (e.g., 'data_ready')
+
+    Returns
+    -------
+    Callable[..., str]
+        A callable that returns the fully namespaced event name (str) when invoked
+
+    Examples
+    --------
+    With @trigger decorator:
+
+        ```
+        from metaflow import namespaced_event_name, project
+
+        @project(name="foo")
+        @trigger(event=namespaced_event_name('data_ready'))
+        class MyFlow(FlowSpec):
+            ...
+
+        @project(name="foo")
+        @trigger(event={'name': namespaced_event_name('data_ready'), 'parameters': {'x': 'y'}})
+        class MyFlow(FlowSpec):
+        ```
+
+    With ArgoEvent:
+
+        ```
+        from metaflow import namespaced_event_name
+        from metaflow.integrations import ArgoEvent
+
+        ArgoEvent(namespaced_event_name('data_ready')).publish()
+        ```
+    """
+    if event_name is None or not is_stringish(event_name):
+        raise ValueError(
+            "event_name should be a string not %s" % (str(type(event_name)))
+        )
+
+    def _delayed_name_generator(context=None, event_name=None):
+        from metaflow import current
+
+        project_name = current.get("project_name", None)
+        branch_name = current.get("branch_name", None)
+        return _make_namespaced_event_name(event_name, project_name, branch_name)
+
+    return functools.partial(_delayed_name_generator, event_name=event_name)
+
+
+def _make_namespaced_event_name(logical_name, project=None, branch=None):
+    """
+    Construct a namespaced event name from components.
+
+    Parameters
+    ----------
+    logical_name : str
+        The base event name
+    project : str, optional
+        Project name from @project decorator
+    branch : str, optional
+        Branch name from @project decorator
+
+    Returns
+    -------
+    str
+        Namespaced event name in format mfns.<project>.<branch>.<logical_name>
+        or mfns.<logical_name> if no project/branch
+    """
+    if project and branch:
+        # When it comes to project and branches we want to make
+        # sure that we are not having any non-acceptable characters.
+        # allow dot (.) in project and branch names
+        branch = re.sub(r"[^a-zA-Z0-9_\-\.]", "", branch)
+        project = re.sub(r"[^a-zA-Z0-9_\-\.]", "", project)
+        return f"{NAMESPACED_EVENTS_PREFIX}.{project}.{branch}.{logical_name}"
+    else:
+        return f"{NAMESPACED_EVENTS_PREFIX}.{logical_name}"