Concise dynamic node ids and names

flytekit/common/utils.py

@@ -11,7 +11,7 @@
 from flytekit.models.core import identifier as _identifier
 
 
-def _dnsify(value):  # type: (str) -> str
+def _dnsify(value: str) -> str:
     """
     Converts value into a DNS-compliant (RFC1035/RFC1123 DNS_LABEL). The resulting string must only consist of
     alphanumeric (lower-case a-z, and 0-9) and not exceed 63 characters. It's permitted to have '-' character as long

flytekit/core/base_task.py

@@ -32,6 +32,7 @@
     VoidPromise,
     create_and_link_node,
     create_task_output,
+    extract_obj_name,
     flyte_entity_call_handler,
     translate_inputs_to_literals,
 )
@@ -417,7 +418,7 @@ def construct_node_metadata(self) -> _workflow_model.NodeMetadata:
         Used when constructing the node that encapsulates this task as part of a broader workflow definition.
         """
         return _workflow_model.NodeMetadata(
-            name=f"{self.__module__}.{self.name}",
+            name=extract_obj_name(self.name),
             timeout=self.metadata.timeout,
             retries=self.metadata.retry_strategy,
             interruptible=self.metadata.interruptible,

flytekit/core/promise.py

@@ -730,6 +730,18 @@ def construct_node_metadata(self) -> _workflow_model.NodeMetadata:
         ...
 
 
+def extract_obj_name(name: str) -> str:
+    """
+    Generates a shortened name, without the module information. Useful for node-names etc. Only extracts the final
+    object information often separated by `.` in the python fully qualified notation
+    """
+    if name is None:
+        return ""
+    if "." in name:
+        return name.split(".")[-1]
+    return name
+
+
 def create_and_link_node(
     ctx: FlyteContext,
     entity: SupportsNodeCreation,

flytekit/core/python_function_task.py

@@ -169,9 +169,9 @@ def compile_into_workflow(
         from flytekit.core.task import ReferenceTask
 
         if not ctx.compilation_state:
-            cs = ctx.new_compilation_state("dynamic")
+            cs = ctx.new_compilation_state(prefix="d")
         else:
-            cs = ctx.compilation_state.with_params(prefix="dynamic")
+            cs = ctx.compilation_state.with_params(prefix="d")
 
         with FlyteContextManager.with_context(ctx.with_compilation_state(cs)):
             # TODO: Resolve circular import

flytekit/core/reference_entity.py

@@ -10,6 +10,7 @@
     VoidPromise,
     create_and_link_node,
     create_task_output,
+    extract_obj_name,
     translate_inputs_to_literals,
 )
 from flytekit.core.type_engine import TypeEngine
@@ -176,7 +177,7 @@ def local_execute(self, ctx: FlyteContext, **kwargs) -> Optional[Union[Tuple[Pro
         return create_task_output(vals, self.python_interface)
 
     def construct_node_metadata(self) -> _workflow_model.NodeMetadata:
-        return _workflow_model.NodeMetadata(name=f"{self.__module__}.{self.name}")
+        return _workflow_model.NodeMetadata(name=extract_obj_name(self.name))
 
     def compile(self, ctx: FlyteContext, *args, **kwargs):
         return create_and_link_node(ctx, entity=self, **kwargs)

flytekit/core/workflow.py

@@ -27,6 +27,7 @@
     VoidPromise,
     binding_from_python_std,
     create_task_output,
+    extract_obj_name,
     flyte_entity_call_handler,
     translate_inputs_to_literals,
 )
@@ -186,7 +187,7 @@ def name(self) -> str:
 
     @property
     def short_name(self) -> str:
-        return self._name.split(".")[-1]
+        return extract_obj_name(self._name)
 
     @property
     def workflow_metadata(self) -> Optional[WorkflowMetadata]:
@@ -222,7 +223,7 @@ def __repr__(self):
 
     def construct_node_metadata(self) -> _workflow_model.NodeMetadata:
         return _workflow_model.NodeMetadata(
-            name=f"{self.__module__}.{self.name}",
+            name=extract_obj_name(self.name),
             interruptible=self.workflow_metadata_defaults.interruptible,
         )
 
@@ -601,7 +602,7 @@ def compile(self, **kwargs):
         ctx = FlyteContextManager.current_context()
         self._input_parameters = transform_inputs_to_parameters(ctx, self.python_interface)
         all_nodes = []
-        prefix = f"{ctx.compilation_state.prefix}-{self.short_name}-" if ctx.compilation_state is not None else ""
+        prefix = ctx.compilation_state.prefix if ctx.compilation_state is not None else ""
 
         with FlyteContextManager.with_context(
             ctx.with_compilation_state(CompilationState(prefix=prefix, task_resolver=self))

tests/flytekit/unit/common_tests/test_promise.py

@@ -5,7 +5,7 @@
 from flytekit.common.exceptions import user as _user_exceptions
 from flytekit.common.types import base_sdk_types, primitives
 from flytekit.core.interface import Interface
-from flytekit.core.promise import Promise, create_native_named_tuple
+from flytekit.core.promise import Promise, create_native_named_tuple, extract_obj_name
 from flytekit.core.type_engine import TypeEngine
 from flytekit.models.types import LiteralType, SimpleType
 
@@ -70,3 +70,20 @@ def test_create_native_named_tuple():
 
     with pytest.raises(AssertionError, match="Failed to convert value of output x"):
         create_native_named_tuple(ctx, promises=p1, entity_interface=Interface(outputs={"x": Promise}))
+
+
+@pytest.mark.parametrize(
+    "name, expected_name",
+    [
+        ("test", "test"),
+        ("test.abc", "abc"),
+        (".test", "test"),
+        ("test.", ""),
+        ("test.xyz.abc", "abc"),
+        ("", ""),
+        (None, ""),
+        ("test.xyz.abc.", ""),
+    ],
+)
+def test_extract_obj_name(name, expected_name):
+    assert extract_obj_name(name) == expected_name

tests/flytekit/unit/common_tests/test_utils.py

@@ -0,0 +1,22 @@
+import pytest
+
+from flytekit.common.utils import _dnsify
+
+
+@pytest.mark.parametrize(
+    "input,expected",
+    [
+        ("test.abc", "test-abc"),
+        ("test", "test"),
+        ("", ""),
+        (".test", "test"),
+        ("Test", "test"),
+        ("test.", "test"),
+        ("test-", "test"),
+        ("test$", "test"),
+        ("te$t$", "tet"),
+        ("t" * 64, f"da4b348ebe-{'t'*52}"),
+    ],
+)
+def test_dnsify(input, expected):
+    assert _dnsify(input) == expected

tests/flytekit/unit/core/test_workflows.py

@@ -191,7 +191,7 @@ def wf2() -> int:
 
     sub_wf = model_wf.sub_workflows[0]
     assert len(sub_wf.nodes) == 1
-    assert sub_wf.nodes[0].id == "wf2-n0"
+    assert sub_wf.nodes[0].id == "n0"
     assert sub_wf.nodes[0].task_node.reference_id.name == "test_workflows.t1"