diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index a3adbe4d6..7cb15f937 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -12,7 +12,6 @@ from collections.abc import Callable, Mapping, Sequence from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime, timezone -from enum import Enum from typing import Protocol, TypeVar import click @@ -30,6 +29,7 @@ from langgraph_cli.host_backend import ( ControlPlaneEndpoints, HostBackendClient, HostBackendError, + SourceName, ) from langgraph_cli.image_reference import ImageReference from langgraph_cli.progress import Progress @@ -109,11 +109,7 @@ _HYBRID_LISTENER_GUIDANCE = ( "then re-run with --deployment-id ." ) - -class SourceId(str, Enum): - INTERNAL_DOCKER = "internal_docker" - INTERNAL_SOURCE = "internal_source" - EXTERNAL_DOCKER = "external_docker" +_CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker" _TERMINAL_STATUSES = frozenset( @@ -1167,7 +1163,7 @@ def _run_local_build( updated = client.update_deployment( deployment_id, resolved_image, - revision_source=SourceId.INTERNAL_DOCKER, + revision_source="internal_docker", secrets=secrets, tracked_packages=tracked_packages, ) @@ -1301,7 +1297,7 @@ class DeploymentSource(Protocol): def _resolve_or_create( - ctx: DeployContext, *, source: SourceId, not_found_message: str + ctx: DeployContext, *, source: SourceName, not_found_message: str ) -> tuple[str, int]: if isinstance(ctx.selector, ById): existing, step = _fetch_deployment(ctx.client, 1, ctx.selector) @@ -1324,7 +1320,7 @@ def _resolve_or_create( def _ensure_customer_registry_source(existing: ExistingDeployment) -> None: - if existing.source != SourceId.EXTERNAL_DOCKER: + if existing.source != _CUSTOMER_REGISTRY_SOURCE: raise click.UsageError( f"Deployment {existing.id} was not created from an external image " "and cannot be updated with --push-to. Run without --push-to to keep " @@ -1342,7 +1338,7 @@ class ManagedRegistrySource: def run(self, ctx: DeployContext) -> DeployOutcome: deployment_id, step = _resolve_or_create( ctx, - source=SourceId.INTERNAL_DOCKER, + source="internal_docker", not_found_message="No deployment found. Will create after build.", ) build_result = _run_local_build( @@ -1366,7 +1362,7 @@ class RemoteBuildSource: def run(self, ctx: DeployContext) -> DeployOutcome: deployment_id, step = _resolve_or_create( ctx, - source=SourceId.INTERNAL_SOURCE, + source="internal_source", not_found_message="No deployment found. Will create.", ) build_result = _run_remote_build( @@ -1424,7 +1420,7 @@ class CustomerRegistrySource: ctx.client, step, name=name, - source=SourceId.EXTERNAL_DOCKER, + source=_CUSTOMER_REGISTRY_SOURCE, source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC}, source_revision_config={"image_uri": image_uri}, secrets=ctx.secrets, diff --git a/libs/cli/langgraph_cli/host_backend.py b/libs/cli/langgraph_cli/host_backend.py index fb3b7e729..ebfd3f9d0 100644 --- a/libs/cli/langgraph_cli/host_backend.py +++ b/libs/cli/langgraph_cli/host_backend.py @@ -3,7 +3,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any +from typing import Any, Literal from urllib.parse import urlparse import click @@ -18,6 +18,7 @@ CLOUD_DASHBOARD_HOST = "smith.langchain.com" CONTROL_PLANE_PATH = "/api-host" LANGSMITH_API_PATHS = ("/api/v1", "/api") LOCAL_HOSTNAMES = ("localhost", "127.0.0.1") +SourceName = Literal["internal_docker", "internal_source", "external_docker"] @dataclass(frozen=True, slots=True) @@ -156,7 +157,7 @@ class HostBackendClient: self, *, name: str, - source: str, + source: SourceName, source_config: dict[str, object], source_revision_config: dict[str, object], secrets: list[dict[str, str]] | None = None, @@ -202,7 +203,7 @@ class HostBackendClient: deployment_id: str, image_uri: str, *, - revision_source: str | None, + revision_source: SourceName | None, secrets: list[dict[str, str]] | None = None, tracked_packages: list[str] | None = None, ) -> dict[str, Any]: