allow for overriding workflow name

CHANGELOG.md

@@ -16,6 +16,12 @@ Types of changes
 
 # Latch SDK Changelog
 
+## 2.75.0 - 2026-05-21
+
+## Added
+
+* `latch register --workflow-name` option to override workflow name for registration.
+
 ## 2.74.0 - 2026-05-18
 
 ## Added

pyproject.toml

@@ -12,7 +12,7 @@ include = ["src/**/*.py", "src/**/*.graphql", "src/**/py.typed", "src/latch_cli/
 
 [project]
 name = "latch"
-version = "2.74.0"
+version = "2.75.0"
 description = "The Latch SDK"
 authors = [{ name = "Kenny Workman", email = "kenny@latch.bio" }]
 maintainers = [{ name = "Kenny Workman", email = "kenny@latch.bio" }]

src/latch/resources/workflow.py

@@ -52,7 +52,11 @@ def workflow(
         if f.__doc__ is None or "__metadata__:" not in f.__doc__:
             metadata = _generate_metadata(f)
             _inject_metadata(f, metadata)
-        return _workflow(f)
+        wf_name_override = os.environ.get("LATCH_WF_NAME_OVERRIDE")
+        if wf_name_override is not None and wf_name_override.strip() == "":
+            wf_name_override = None
+
+        return _workflow(f, wf_name_override=wf_name_override)
 
     def decorator(f: Callable):
         signature = inspect.signature(f)

src/latch_cli/centromere/ctx.py

@@ -1,8 +1,8 @@
-from logging import getLogger
 import re
 import sys
 import traceback
 from dataclasses import dataclass
+from logging import getLogger
 from pathlib import Path
 from textwrap import dedent
 from typing import Dict, Optional, Tuple
@@ -94,6 +94,7 @@ def __init__(
         use_new_centromere: bool = False,
         overwrite: bool = False,
         dockerfile_path: Optional[Path] = None,
+        explicit_workflow_name: Optional[str] = None,
     ):
         self.use_new_centromere = use_new_centromere
         self.remote = remote
@@ -242,7 +243,11 @@ def __init__(
                     )
                     raise click.exceptions.Exit(1)
 
-                self.workflow_name = wf_name
+                self.workflow_name = (
+                    explicit_workflow_name
+                    if explicit_workflow_name is not None
+                    else wf_name
+                )
 
                 for obj in flyte_objects:
                     if obj.type != "task" or obj.dockerfile is None:
@@ -373,8 +378,6 @@ def __init__(
                     )
                     raise click.exceptions.Exit(1)
 
-                # todo(kenny): support per container task and custom workflow
-                # name for snakemake
                 self.workflow_name = f"{metadata._snakemake_metadata.name}_jit_register"
             else:
                 assert self.nf_script is not None
@@ -408,6 +411,9 @@ def __init__(
                 if name_path.exists():
                     self.workflow_name = name_path.read_text().strip()
 
+            if explicit_workflow_name is not None:
+                self.workflow_name = explicit_workflow_name
+
             assert self.workflow_name is not None
 
             if self.nucleus_check_version(self.version, self.workflow_name):

src/latch_cli/main.py

@@ -33,6 +33,7 @@
     get_latest_package_version,
     get_local_package_version,
     hash_directory,
+    normalize_explicit_workflow_name,
 )
 from latch_cli.workflow_config import BaseImageOptions
 from latch_sdk_gql.execute import execute as gql_execute
@@ -629,6 +630,21 @@ def image_ls():
     ls()
 
 
+def _validate_explicit_workflow_name(
+    ctx: click.Context,
+    _param: click.Parameter,
+    value: Optional[str],
+) -> Optional[str]:
+    if value is None:
+        return None
+
+    value = normalize_explicit_workflow_name(value)
+    if value is None:
+        ctx.fail("--workflow-name must not be empty.")
+
+    return value
+
+
 @latch.command("register")
 @click.argument("pkg_root", type=click.Path(exists=True, file_okay=False))
 @click.option(
@@ -689,6 +705,14 @@ def image_ls():
     type=str,
     help="Module containing Latch workflow to register. Defaults to `wf`",
 )
+@click.option(
+    "--workflow-name",
+    "explicit_workflow_name",
+    type=str,
+    default=None,
+    callback=_validate_explicit_workflow_name,
+    help="Override workflow name to use for this registration.",
+)
 @click.option(
     "--metadata-root",
     type=click.Path(exists=False, path_type=Path, file_okay=False),
@@ -762,6 +786,7 @@ def register(
     yes: bool,
     open: bool,
     workflow_module: Optional[str],
+    explicit_workflow_name: Optional[str],
     metadata_root: Optional[Path],
     snakefile: Optional[Path],
     cache_tasks: bool,
@@ -791,6 +816,7 @@ def register(
             remote=remote,
             skip_confirmation=yes,
             wf_module=workflow_module,
+            explicit_workflow_name=explicit_workflow_name,
             progress_plain=(docker_progress == "auto" and not sys.stdout.isatty())
             or docker_progress == "plain",
             dockerfile_path=dockerfile,
@@ -828,6 +854,7 @@ def register(
         skip_confirmation=yes,
         open=open,
         wf_module=workflow_module,
+        explicit_workflow_name=explicit_workflow_name,
         metadata_root=metadata_root,
         snakefile=snakefile,
         nf_script=nf_script,

src/latch_cli/nextflow/workflow.py

@@ -396,12 +396,15 @@ def generate_nextflow_workflow(
     dest: Path,
     *,
     execution_profile: Optional[str] = None,
+    workflow_name: Optional[str] = None,
 ):
     generate_nextflow_config(pkg_root)
 
     assert metadata._nextflow_metadata is not None
 
-    wf_name = metadata._nextflow_metadata.name
+    wf_name = workflow_name
+    if wf_name is None:
+        wf_name = metadata._nextflow_metadata.name
     assert wf_name is not None
 
     parameters = metadata._nextflow_metadata.parameters

src/latch_cli/services/register/register.py

@@ -28,7 +28,7 @@
     serialize_pkg_in_container,
     upload_image,
 )
-from latch_cli.utils import WorkflowType
+from latch_cli.utils import WorkflowType, normalize_explicit_workflow_name
 
 log = getLogger(__name__)
 
@@ -311,6 +311,7 @@ def register(
     open: bool = False,
     skip_confirmation: bool = False,
     wf_module: Optional[str] = None,
+    explicit_workflow_name: Optional[str] = None,
     metadata_root: Optional[Path] = None,
     snakefile: Optional[Path] = None,
     nf_script: Optional[Path] = None,
@@ -374,6 +375,9 @@ def register(
             click.secho("\n`snakemake` package is not installed.", fg="red", bold=True)
             sys.exit(1)
 
+    explicit_workflow_name = normalize_explicit_workflow_name(explicit_workflow_name)
+    use_explicit_workflow_name = explicit_workflow_name is not None
+
     with _CentromereCtx(
         Path(pkg_root),
         disable_auto_version=disable_auto_version,
@@ -385,6 +389,7 @@ def register(
         use_new_centromere=use_new_centromere,
         overwrite=skip_confirmation,
         dockerfile_path=dockerfile_path,
+        explicit_workflow_name=explicit_workflow_name,
     ) as ctx:
         assert ctx.workflow_name is not None, "Unable to determine workflow name"
         assert ctx.version is not None, "Unable to determine workflow version"
@@ -397,6 +402,10 @@ def register(
             ])
         )
         click.echo(" ".join([click.style("Version:", fg="bright_blue"), ctx.version]))
+        if use_explicit_workflow_name:
+            click.echo(
+                "Using workflow name from --workflow-name; .latch/workflow_name will not be updated."
+            )
 
         if workspace_id is None:
             workspace_id = current_workspace()
@@ -467,7 +476,10 @@ def register(
             from ...snakemake.workflow import build_jit_register_wrapper
 
             sm_jit_wf = build_jit_register_wrapper(
-                cache_tasks, ctx.git_commit_hash, ctx.git_is_dirty
+                cache_tasks,
+                ctx.git_commit_hash,
+                ctx.git_is_dirty,
+                workflow_name=ctx.workflow_name,
             )
             generate_jit_register_code(
                 sm_jit_wf,
@@ -492,6 +504,7 @@ def register(
                 ctx.nf_script,
                 dest,
                 execution_profile=nf_execution_profile,
+                workflow_name=ctx.workflow_name,
             )
 
         click.secho("\nInitializing registration", bold=True)
@@ -614,7 +627,8 @@ def register(
             wf_name = ctx.workflow_name
 
             name_path = Path(pkg_root) / latch_constants.pkg_workflow_name
-            if not name_path.exists():
+            if not use_explicit_workflow_name and not name_path.exists():
+                name_path.parent.mkdir(parents=True, exist_ok=True)
                 name_path.write_text(ctx.workflow_name)
 
             while len(wf_infos) == 0:

src/latch_cli/services/register/staging.py

@@ -11,7 +11,12 @@
 
 from ...centromere.ast_parsing import get_flyte_objects
 from ...constants import docker_image_name_illegal_pat, latch_constants
-from ...utils import WorkflowType, hash_directory, identifier_suffix_from_str
+from ...utils import (
+    WorkflowType,
+    hash_directory,
+    identifier_suffix_from_str,
+    normalize_explicit_workflow_name,
+)
 from ..docker.utils import dbnp, get_local_docker_client, remote_dbnp
 
 
@@ -23,6 +28,7 @@ def register_staging(
     remote: bool = False,
     skip_confirmation: bool = False,
     wf_module: Optional[str] = None,
+    explicit_workflow_name: Optional[str] = None,
     progress_plain: bool = False,
     dockerfile_path: Optional[Path] = None,
 ):
@@ -50,10 +56,10 @@ def register_staging(
         )
         raise click.exceptions.Exit(1) from e
 
-    wf_name: Optional[str] = None
+    wf_name: Optional[str] = normalize_explicit_workflow_name(explicit_workflow_name)
 
     name_path = pkg_root / latch_constants.pkg_workflow_name
-    if name_path.exists():
+    if wf_name is None and name_path.exists():
         click.echo(f"Parsing workflow name from {name_path}.")
         wf_name = name_path.read_text().strip()
 

src/latch_cli/snakemake/workflow.py

@@ -316,6 +316,7 @@ def __init__(
         cache_tasks: bool = False,
         git_commit_hash: Optional[str] = None,
         git_is_dirty: bool = False,
+        workflow_name: Optional[str] = None,
     ):
         self.cache_tasks = cache_tasks
 
@@ -367,7 +368,7 @@ def __init__(
         workflow_metadata = WorkflowMetadata(
             on_failure=WorkflowFailurePolicy.FAIL_IMMEDIATELY
         )
-        name = f"{name}_jit_register"
+        name = workflow_name if workflow_name is not None else f"{name}_jit_register"
         workflow_metadata_defaults = WorkflowMetadataDefaults(False)
         super().__init__(
             name=name,
@@ -577,8 +578,9 @@ def get_fn_code(
         remote_output_url: Optional[str],
     ):
         task_name = f"{self.name}_task"
+        fn_name = identifier_from_str(task_name)
 
-        code_block = self.get_fn_interface(fn_name=task_name)
+        code_block = self.get_fn_interface(fn_name=fn_name)
 
         code_block += reindent(
             rf"""
@@ -763,6 +765,11 @@ class _WorkflowInfoNode(TypedDict):
             1,
         )
         code_block += self.get_fn_return_stmt()
+        code_block += dedent(rf"""
+
+            {fn_name}._name = {task_name!r}
+            setattr(sys.modules[__name__], {task_name!r}, {fn_name})
+            """)
         return code_block
 
 
@@ -1032,8 +1039,11 @@ def build_jit_register_wrapper(
     cache_tasks: bool = False,
     git_commit_hash: Optional[str] = None,
     git_is_dirty: bool = False,
+    workflow_name: Optional[str] = None,
 ) -> JITRegisterWorkflow:
-    wrapper_wf = JITRegisterWorkflow(cache_tasks, git_commit_hash, git_is_dirty)
+    wrapper_wf = JITRegisterWorkflow(
+        cache_tasks, git_commit_hash, git_is_dirty, workflow_name
+    )
     out_parameter_name = wrapper_wf.out_parameter_name
 
     python_interface = wrapper_wf.python_interface

src/latch_cli/utils/__init__.py

@@ -13,7 +13,7 @@
 from logging import getLogger
 from pathlib import Path
 from textwrap import dedent
-from typing import List
+from typing import List, Optional
 from urllib.parse import urljoin
 
 import click
@@ -70,6 +70,17 @@ def urljoins(*args: str, dir: bool = False) -> str:
 class AuthenticationError(RuntimeError): ...
 
 
+def normalize_explicit_workflow_name(workflow_name: Optional[str]) -> Optional[str]:
+    if workflow_name is None:
+        return None
+
+    workflow_name = workflow_name.strip()
+    if len(workflow_name) == 0:
+        return None
+
+    return workflow_name
+
+
 def get_auth_header() -> str:
     sdk_token = user_config.token
     execution_token = os.environ.get("FLYTE_INTERNAL_EXECUTION_ID")