Add flyte.app.Subdomain for deploy-time subdomain resolution

Subdomain.from_app_name(app_name, project_domain_suffix="hash"|"default")
produces {app_name}-{hash-of-project-domain} or {app_name}-{project}-{domain},
and Subdomain.from_function(fn) lets users compute the subdomain from the
AppEnvironment and the deployment SerializationContext. Domain.subdomain now
accepts a str or a Subdomain, resolved during app serialization.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KVnRzHRUANmCg7ZsLH6Fah

Author: cosmicBboy

src/flyte/app/__init__.py

@@ -3,7 +3,7 @@
 from flyte.app._context import ctx
 from flyte.app._deploy import DeployedAppEnvironment
 from flyte.app._parameter import AppEndpoint, ArtifactValue, Parameter, RunOutput, get_parameter
-from flyte.app._types import Domain, Link, Port, Scaling, Timeouts
+from flyte.app._types import Domain, Link, Port, Scaling, Subdomain, Timeouts
 
 __all__ = [
     "AppEndpoint",
@@ -17,6 +17,7 @@
     "Port",
     "RunOutput",
     "Scaling",
+    "Subdomain",
     "Timeouts",
     "ctx",
     "get_parameter",

src/flyte/app/_runtime/app_serde.py

@@ -22,7 +22,7 @@
 from flyte._internal.runtime.resources_serde import get_proto_extended_resources, get_proto_resources
 from flyte._internal.runtime.task_serde import get_security_context, lookup_image_in_cache
 from flyte._logging import logger
-from flyte.app import AppEnvironment, Parameter, Scaling
+from flyte.app import AppEnvironment, Parameter, Scaling, Subdomain
 from flyte.app._parameter import AppEndpoint, ArtifactValue, _DelayedValue
 from flyte.models import SerializationContext
 from flyte.syncify import syncify
@@ -407,9 +407,12 @@ async def translate_app_env_to_idl(
         msg = "image must be a str, Image, or PodTemplate"
         raise ValueError(msg)
 
+    subdomain = app_env.domain.subdomain if app_env.domain else None
+    if isinstance(subdomain, Subdomain):
+        subdomain = subdomain.resolve(app_env, serialization_context)
     ingress = app_definition_pb2.IngressConfig(
         private=False,
-        subdomain=app_env.domain.subdomain if app_env.domain else None,
+        subdomain=subdomain,
         cname=app_env.domain.custom_domain if app_env.domain else None,
     )
 

src/flyte/app/_types.py

@@ -1,6 +1,11 @@
+import hashlib
 from dataclasses import dataclass
 from datetime import timedelta
-from typing import Optional, Tuple, Union
+from typing import TYPE_CHECKING, Callable, Literal, Optional, Tuple, Union
+
+if TYPE_CHECKING:
+    from flyte.app._app_environment import AppEnvironment
+    from flyte.models import SerializationContext
 
 import rich.repr
 
@@ -150,14 +155,91 @@ def __post_init__(self):
             raise ValueError("request timeout must not exceed 1 hour (3600 seconds)")
 
 
+_PROJECT_DOMAIN_HASH_LEN = 8
+
+
+@rich.repr.auto
+@dataclass(frozen=True)
+class Subdomain:
+    """
+    A subdomain that is resolved at deploy time, when the deployment project and domain are known.
+
+    Use `Subdomain.from_app_name` for the built-in naming schemes:
+
+    - `project_domain_suffix="hash"`: the subdomain is `{app_name}-{hash}`, where the hash is computed
+      from `{project}-{domain}`. This keeps subdomains short and stable per project/domain.
+    - `project_domain_suffix="default"`: the subdomain is `{app_name}-{project}-{domain}`.
+
+    Use `Subdomain.from_function` for full control: the function receives the `AppEnvironment` and the
+    deployment `SerializationContext` (project, domain, org, version, ...) and returns the subdomain.
+
+    The final subdomain string is produced by `resolve()` during serialization.
+    """
+
+    app_name: Optional[str] = None
+    project_domain_suffix: Literal["hash", "default"] = "hash"
+    function: Optional[Callable[["AppEnvironment", "SerializationContext"], str]] = None
+
+    def __post_init__(self):
+        if (self.app_name is None) == (self.function is None):
+            raise ValueError("exactly one of app_name or function must be set")
+        if self.project_domain_suffix not in ("hash", "default"):
+            raise ValueError(f"project_domain_suffix must be 'hash' or 'default', got {self.project_domain_suffix!r}")
+
+    @classmethod
+    def from_app_name(cls, app_name: str, project_domain_suffix: Literal["hash", "default"] = "hash") -> "Subdomain":
+        """
+        Create a subdomain for an app whose final value depends on the deployment project and domain.
+
+        Args:
+            app_name: Name of the app.
+            project_domain_suffix: `"hash"` for `{app_name}-{hash-of-project-domain}`, or `"default"`
+                for `{app_name}-{project}-{domain}`.
+        """
+        return cls(app_name=app_name, project_domain_suffix=project_domain_suffix)
+
+    @classmethod
+    def from_function(cls, function: Callable[["AppEnvironment", "SerializationContext"], str]) -> "Subdomain":
+        """
+        Create a subdomain computed by a user-provided function at deploy time.
+
+        Args:
+            function: Called with the `AppEnvironment` being deployed and the deployment
+                `SerializationContext`; returns the subdomain string.
+        """
+        return cls(function=function)
+
+    def resolve(self, app_env: "AppEnvironment", serialization_context: "SerializationContext") -> str:
+        """Resolve to the final subdomain string for the given app environment and deployment context."""
+        if self.function is not None:
+            subdomain = self.function(app_env, serialization_context)
+            if not isinstance(subdomain, str) or not subdomain:
+                raise ValueError(
+                    f"subdomain function for app {app_env.name!r} must return a non-empty str, got {subdomain!r}"
+                )
+            return subdomain
+
+        project, domain = serialization_context.project, serialization_context.domain
+        if not project or not domain:
+            raise ValueError(
+                f"project and domain are required to resolve subdomain for app {self.app_name!r}, "
+                f"got project={project!r}, domain={domain!r}"
+            )
+        if self.project_domain_suffix == "hash":
+            suffix = hashlib.sha256(f"{project}-{domain}".encode()).hexdigest()[:_PROJECT_DOMAIN_HASH_LEN]
+            return f"{self.app_name}-{suffix}"
+        return f"{self.app_name}-{project}-{domain}"
+
+
 @rich.repr.auto
 @dataclass
 class Domain:
     # SubDomain config
 
-    """Subdomain to use for the domain. If not set, the default subdomain will be used."""
+    """Subdomain to use for the domain. Either a literal string, or a `Subdomain` resolved against the
+    deployment project and domain. If not set, the default subdomain will be used."""
 
-    subdomain: Optional[str] = None
+    subdomain: Optional[Union[str, Subdomain]] = None
 
     """Custom domain to use for the domain. If not set, the default custom domain will be used."""
     custom_domain: Optional[str] = None

tests/flyte/app/runtime/test_app_serde.py

@@ -5,6 +5,7 @@
 into protobuf IDL format without using mocks.
 """
 
+import hashlib
 import pathlib
 from datetime import timedelta
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -28,7 +29,7 @@
     translate_app_env_to_idl,
     translate_parameters,
 )
-from flyte.app._types import Domain, Port, Scaling, Timeouts
+from flyte.app._types import Domain, Port, Scaling, Subdomain, Timeouts
 from flyte.models import CodeBundle, SerializationContext
 
 
@@ -1399,3 +1400,42 @@ async def test_translate_parameters_without_artifact_ids_is_unchanged():
 
     assert inputs.items[0].WhichOneof("value") == "string_value"
     assert inputs.items[0].string_value == "s3://bucket/weights.pt"
+
+
+@pytest.mark.parametrize(
+    "subdomain,expected",
+    [
+        (
+            Subdomain.from_app_name("test-app"),
+            "test-app-" + hashlib.sha256(b"test-project-test-domain").hexdigest()[:8],
+        ),
+        (
+            Subdomain.from_app_name("test-app", project_domain_suffix="default"),
+            "test-app-test-project-test-domain",
+        ),
+        (
+            Subdomain.from_function(lambda app_env, ctx: f"{app_env.name}-{ctx.org}"),
+            "test-app-test-org",
+        ),
+    ],
+)
+def test_app_with_resolved_subdomain(subdomain: Subdomain, expected: str):
+    """
+    GOAL: Verify Subdomain instances are resolved to the final subdomain string in the ingress config.
+    """
+    app_env = AppEnvironment(
+        name="test-app",
+        image=Image.from_base("python:3.11"),
+        domain=Domain(subdomain=subdomain),
+    )
+
+    ctx = SerializationContext(
+        org="test-org",
+        project="test-project",
+        domain="test-domain",
+        version="v1",
+        root_dir=pathlib.Path.cwd(),
+    )
+
+    app_idl = translate_app_env_to_idl(app_env, ctx)
+    assert app_idl.spec.ingress.subdomain == expected

tests/flyte/app/test_types.py

@@ -0,0 +1,83 @@
+"""Tests for flyte.app._types."""
+
+import hashlib
+import pathlib
+
+import pytest
+
+from flyte._image import Image
+from flyte.app import AppEnvironment
+from flyte.app._types import Domain, Subdomain
+from flyte.models import SerializationContext
+
+
+def _ctx(project="my-project", domain="my-domain") -> SerializationContext:
+    return SerializationContext(
+        org="my-org",
+        project=project,
+        domain=domain,
+        version="v1",
+        root_dir=pathlib.Path.cwd(),
+    )
+
+
+def _app_env(name="my-app") -> AppEnvironment:
+    return AppEnvironment(name=name, image=Image.from_base("python:3.11"))
+
+
+def test_subdomain_from_app_name_hash():
+    subdomain = Subdomain.from_app_name("my-app")
+    expected_hash = hashlib.sha256(b"my-project-my-domain").hexdigest()[:8]
+    assert subdomain.resolve(_app_env(), _ctx()) == f"my-app-{expected_hash}"
+
+
+def test_subdomain_hash_is_stable_per_project_domain():
+    ctx = _ctx()
+    resolved_a = Subdomain.from_app_name("app-a").resolve(_app_env("app-a"), ctx)
+    resolved_b = Subdomain.from_app_name("app-b").resolve(_app_env("app-b"), ctx)
+    # Same project/domain -> same hash suffix
+    assert resolved_a.split("app-a-")[1] == resolved_b.split("app-b-")[1]
+    # Different project -> different hash suffix
+    other = Subdomain.from_app_name("app-a").resolve(_app_env("app-a"), _ctx(project="other"))
+    assert other != resolved_a
+
+
+def test_subdomain_from_app_name_default():
+    subdomain = Subdomain.from_app_name("my-app", project_domain_suffix="default")
+    assert subdomain.resolve(_app_env(), _ctx()) == "my-app-my-project-my-domain"
+
+
+def test_subdomain_from_function():
+    subdomain = Subdomain.from_function(lambda app_env, ctx: f"{app_env.name}.{ctx.org}.{ctx.domain}")
+    assert subdomain.resolve(_app_env(), _ctx()) == "my-app.my-org.my-domain"
+
+
+def test_subdomain_from_function_invalid_return():
+    subdomain = Subdomain.from_function(lambda app_env, ctx: None)
+    with pytest.raises(ValueError, match="non-empty str"):
+        subdomain.resolve(_app_env(), _ctx())
+
+
+def test_subdomain_invalid_suffix():
+    with pytest.raises(ValueError, match="project_domain_suffix"):
+        Subdomain.from_app_name("my-app", project_domain_suffix="bogus")
+
+
+def test_subdomain_requires_app_name_or_function():
+    with pytest.raises(ValueError, match="exactly one"):
+        Subdomain()
+    with pytest.raises(ValueError, match="exactly one"):
+        Subdomain(app_name="my-app", function=lambda app_env, ctx: "x")
+
+
+def test_subdomain_resolve_requires_project_and_domain():
+    subdomain = Subdomain.from_app_name("my-app")
+    with pytest.raises(ValueError, match="project and domain are required"):
+        subdomain.resolve(_app_env(), _ctx(project=None))
+
+
+def test_domain_accepts_subdomain():
+    domain = Domain(subdomain=Subdomain.from_app_name("my-app"))
+    assert isinstance(domain.subdomain, Subdomain)
+    domain = Domain(subdomain="literal-subdomain")
+    assert domain.subdomain == "literal-subdomain"