WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-27 20:15:00 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(cli): Add deploy revisions list command (#7142)

Browse Source 
### Summary
```
% langgraph deploy revisions list f5bb6e58-e2ab-40b9-9855-d6dd6d1051b6
No LangSmith API key found. Create one at Settings > API Keys in LangSmith.
Enter LangSmith API key: 
Revision ID                           Status    Created At                 
------------------------------------  --------  ---------------------------
fe05f163-58cf-4783-a918-0c78a8fab0c7  BUILDING  2026-03-12T20:47:14.626436Z
03c32fcf-544f-4517-a016-c7a7243b898d  DEPLOYED  2026-03-12T20:30:09.199275Z
29605810-29cc-4110-ac05-d43e564827d0  DEPLOYED  2026-03-12T19:56:36.232427Z
```
This commit is contained in:
Andrew Nguonly
2026-03-19 16:55:19 -07:00
committed by GitHub
parent 54c92757f2
commit cc9acc1ba9
4 changed files with 283 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/cli/langgraph_cli/cli.py
+83 -22
View File
@@ -31,7 +31,11 @@ from langgraph_cli.helpers import format_log_entry, level_fg, resolve_deployment
from langgraph_cli.host_backend import HostBackendClient, HostBackendError
from langgraph_cli.progress import Progress
from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new
from langgraph_cli.util import format_deployments_table, warn_non_wolfi_distro
from langgraph_cli.util import (
format_deployments_table,
format_revisions_table,
warn_non_wolfi_distro,
)
from langgraph_cli.version import __version__
RESERVED_ENV_VARS = frozenset(
@@ -331,25 +335,25 @@ class NestedHelpGroup(click.Group):
self, ctx: click.Context, formatter: click.HelpFormatter
) -> None:
command_entries: list[tuple[str, click.Command]] = []
# Collect the top-level commands first, then append one level of nested
# subcommands using names like "deploy list" so they show up in the
# top-level help output.
for command_name in self.list_commands(ctx):
command = self.get_command(ctx, command_name)
if command is None or command.hidden:
continue
command_entries.append((command_name, command))
if isinstance(command, click.Group):
# Build a child context so Click resolves the subcommands the same
# way it would for the nested group itself.
sub_ctx = click.Context(command, info_name=command_name, parent=ctx)
for subcommand_name in command.list_commands(sub_ctx):
subcommand = command.get_command(sub_ctx, subcommand_name)
if subcommand is None or subcommand.hidden:
continue
command_entries.append(
(f"{command_name} {subcommand_name}", subcommand)
def collect_commands(
group: click.Group, parent_ctx: click.Context, prefix: str = ""
) -> None:
for command_name in group.list_commands(parent_ctx):
command = group.get_command(parent_ctx, command_name)
if command is None or command.hidden:
continue
qualified_name = f"{prefix} {command_name}" if prefix else command_name
command_entries.append((qualified_name, command))
if isinstance(command, click.Group):
sub_ctx = click.Context(
command,
info_name=qualified_name,
parent=parent_ctx,
)
collect_commands(command, sub_ctx, qualified_name)
collect_commands(self, ctx)
# Compute the available width for help text up front so we can truncate
# descriptions before handing them to Click. That keeps each command on
@@ -373,13 +377,30 @@ class DeployGroup(NestedHelpGroup):
"""Group that treats leading '-' args as passthrough docker flags."""
def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
"""Treat leading option-like subcommand tokens as passthrough args.
Click stores the unresolved nested command token on the context after
``Group.parse_args()`` runs, but the backing attribute changed across
supported Click versions. Click 8.1.x stores the value directly on
``protected_args``, while Click 8.2+ stores it on ``_protected_args``
and exposes ``protected_args`` as a deprecated compatibility property.
Since this package allows ``click>=8.1.7``, we need to check both
names to support the full version range without relying on one
version-specific internal detail.
"""
result = super().parse_args(ctx, args)
if ctx._protected_args and ctx._protected_args[0].startswith("-"):
protected_args = ctx.__dict__.get("protected_args")
if protected_args is None:
protected_args = ctx.__dict__.get("_protected_args", [])
if protected_args and protected_args[0].startswith("-"):
# Click stores the would-be subcommand in _protected_args; if it looks
# like an option (e.g. --build-arg) treat it as passthrough docker
# args instead of insisting on a nested command.
ctx.args = [*ctx._protected_args, *ctx.args]
ctx._protected_args = []
ctx.args = [*protected_args, *ctx.args]
if "protected_args" in ctx.__dict__:
ctx.protected_args = []
elif "_protected_args" in ctx.__dict__:
ctx._protected_args = []
return ctx.args
return result
@@ -788,6 +809,13 @@ def deploy(ctx: click.Context, **_: object):
return ctx.forward(_deploy, docker_build_args=docker_build_args)
@deploy.group(
"revisions", cls=NestedHelpGroup, help="[Beta] Manage deployment revisions."
)
def deploy_revisions() -> None:
pass
@_deploy_base_options()
@click.command(context_settings=dict(ignore_unknown_options=True))
def _deploy(
@@ -1225,6 +1253,39 @@ def deploy_list(api_key: str | None, host_url: str | None, name_contains: str) -
click.echo(format_deployments_table(deployments))
@OPT_HOST_API_KEY
@OPT_HOST_URL
@click.option(
"--limit",
type=int,
default=10,
show_default=True,
help="Maximum number of revisions to return.",
)
@click.argument("deployment_id")
@deploy_revisions.command(
"list",
help=(
"[Beta] List revisions for a LangSmith Deployment.\n\n"
"Use the `deploy list` command to list deployment IDs."
),
)
def deploy_revisions_list(
api_key: str | None, host_url: str | None, limit: int, deployment_id: str
) -> None:
client = _create_host_backend_client(host_url, api_key)
response = _call_host_backend_with_optional_tenant(
client,
lambda c: c.list_revisions(deployment_id, limit=limit),
)
resources = response.get("resources", []) if isinstance(response, dict) else []
revisions = [item for item in resources if isinstance(item, dict)]
if not revisions:
click.echo(f"No revisions found for deployment {deployment_id}.")
return
click.echo(format_revisions_table(revisions))
@OPT_HOST_API_KEY
@OPT_HOST_URL
@click.option(
libs/cli/langgraph_cli/util.py
+32
View File
@@ -57,3 +57,35 @@ def format_deployments_table(deployments: Sequence[dict[str, object]]) -> str:
lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
lines.extend(format_row(row) for row in rows)
return "\n".join(lines)
def format_revisions_table(revisions: Sequence[dict[str, object]]) -> str:
headers = ("Revision ID", "Status", "Created At")
latest_deployed_seen = False
rows = []
for revision in revisions:
status = str(revision.get("status", "-") or "-")
if status == "DEPLOYED":
if latest_deployed_seen:
status = "REPLACED"
else:
latest_deployed_seen = True
rows.append(
(
str(revision.get("id", "-") or "-"),
status,
str(revision.get("created_at", "-") or "-"),
)
)
widths = [
max(len(headers[index]), *(len(row[index]) for row in rows))
for index in range(len(headers))
]
def format_row(row: Sequence[str]) -> str:
return " ".join(value.ljust(widths[index]) for index, value in enumerate(row))
lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
lines.extend(format_row(row) for row in rows)
return "\n".join(lines)
libs/cli/tests/unit_tests/cli/test_cli.py
+136
View File
@@ -297,6 +297,8 @@ def test_top_level_help_shows_deploy_subcommands() -> None:
assert "deploy" in result.output
assert "deploy list" in result.output
assert "deploy delete" in result.output
assert "deploy revisions" in result.output
assert "deploy revisions list" in result.output
assert "[Beta] List LangSmith Deployments." in result.output
@@ -402,6 +404,140 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
assert result.output.strip() == "No deployments found."
def test_deploy_revisions_list_command(monkeypatch) -> None:
runner = CliRunner()
captured: dict[str, str] = {}
class FakeClient:
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
captured["host_url"] = host_url
captured["api_key"] = api_key
captured["tenant_id"] = tenant_id or ""
def list_revisions(self, deployment_id: str, limit: int = 1):
captured["deployment_id"] = deployment_id
captured["limit"] = str(limit)
return {
"resources": [
{
"id": "rev-123",
"status": "CREATING",
"created_at": "2023-11-07T05:31:56Z",
},
{
"id": "rev-456",
"status": "DEPLOYED",
"created_at": "2023-11-08T10:00:00Z",
},
]
}
monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
result = runner.invoke(
cli,
[
"deploy",
"revisions",
"list",
"--api-key",
"test-key",
"--host-url",
"https://api.example.com",
"dep-123",
],
)
assert result.exit_code == 0, result.output
assert captured == {
"host_url": "https://api.example.com",
"api_key": "test-key",
"tenant_id": "",
"deployment_id": "dep-123",
"limit": "10",
}
assert "Revision ID" in result.output
assert "Status" in result.output
assert "Created At" in result.output
assert "rev-123" in result.output
assert "2023-11-08T10:00:00Z" in result.output
def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
runner = CliRunner()
class FakeClient:
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
pass
def list_revisions(self, deployment_id: str, limit: int = 1):
return {"resources": []}
monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
result = runner.invoke(
cli,
[
"deploy",
"revisions",
"list",
"--api-key",
"test-key",
"--host-url",
"https://api.example.com",
"dep-123",
],
)
assert result.exit_code == 0, result.output
assert result.output.strip() == "No revisions found for deployment dep-123."
def test_deploy_revisions_list_command_with_explicit_limit(monkeypatch) -> None:
runner = CliRunner()
captured: dict[str, str] = {}
class FakeClient:
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
pass
def list_revisions(self, deployment_id: str, limit: int = 1):
captured["deployment_id"] = deployment_id
captured["limit"] = str(limit)
return {"resources": []}
monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
result = runner.invoke(
cli,
[
"deploy",
"revisions",
"list",
"--api-key",
"test-key",
"--host-url",
"https://api.example.com",
"--limit",
"25",
"dep-123",
],
)
assert result.exit_code == 0, result.output
assert captured == {"deployment_id": "dep-123", "limit": "25"}
assert result.output.strip() == "No revisions found for deployment dep-123."
def test_deploy_revisions_list_missing_deployment_id_shows_usage() -> None:
runner = CliRunner()
result = runner.invoke(cli, ["deploy", "revisions", "list"])
assert result.exit_code == 2, result.output
assert "Missing argument 'DEPLOYMENT_ID'" in result.output
def test_deploy_delete_command(monkeypatch) -> None:
runner = CliRunner()
captured: dict[str, str] = {}
libs/cli/tests/unit_tests/test_util.py
+32
View File
@@ -4,6 +4,7 @@ from langgraph_cli.util import (
_extract_deployment_url,
clean_empty_lines,
format_deployments_table,
format_revisions_table,
warn_non_wolfi_distro,
)
@@ -224,3 +225,34 @@ def test_format_deployments_table():
assert "alpha" in output
assert "https://alpha.example.com" in output
assert "dep-456" in output
def test_format_revisions_table():
output = format_revisions_table(
[
{
"id": "rev-123",
"status": "DEPLOYED",
"created_at": "2023-11-09T10:00:00Z",
},
{
"id": "rev-456",
"status": "CREATING",
"created_at": "2023-11-07T05:31:56Z",
},
{
"id": "rev-789",
"status": "DEPLOYED",
"created_at": "2023-11-08T10:00:00Z",
},
]
)
assert "Revision ID" in output
assert "Status" in output
assert "Created At" in output
assert "rev-123" in output
assert "CREATING" in output
assert "2023-11-07T05:31:56Z" in output
assert "rev-456" in output
assert "rev-789" in output
assert "REPLACED" in output