diff --git a/libs/cli/langgraph_cli/cli.py b/libs/cli/langgraph_cli/cli.py index 254643a03..fe672240e 100644 --- a/libs/cli/langgraph_cli/cli.py +++ b/libs/cli/langgraph_cli/cli.py @@ -90,27 +90,62 @@ _API_KEY_ENV_NAMES = ( _DEPLOYMENT_NAME_ENV = "LANGSMITH_DEPLOYMENT_NAME" +def _resolve_env_file_path( + config_json: dict, config_path: pathlib.Path +) -> pathlib.Path | None: + """Return the .env file path from config, or None if env is an inline dict.""" + env_field = config_json.get("env") + if isinstance(env_field, dict) and env_field: + return None + if isinstance(env_field, str): + return (config_path.parent / env_field).resolve() + return pathlib.Path.cwd() / ".env" + + def _parse_env_from_config( config_json: dict, config_path: pathlib.Path ) -> dict[str, str]: """Resolve env vars from langgraph.json 'env' field or a .env fallback.""" - env_field = config_json.get("env") - # validate_config_file will default env to {} - if isinstance(env_field, dict) and env_field: - return {str(k): str(v) for k, v in env_field.items()} - if isinstance(env_field, str): - env_path = (config_path.parent / env_field).resolve() - if not env_path.exists(): + env_path = _resolve_env_file_path(config_json, config_path) + if env_path is None: + env_field = config_json.get("env") + if isinstance(env_field, dict) and env_field: + return {str(k): str(v) for k, v in env_field.items()} + return {} + if not env_path.exists(): + env_field = config_json.get("env") + if isinstance(env_field, str): click.secho( f"Warning: env file '{env_field}' specified in langgraph.json not found.", fg="yellow", ) - return {} - else: - env_path = pathlib.Path.cwd() / ".env" + return {} return {k: v for k, v in dotenv_values(env_path).items() if v is not None} +def _save_to_dotenv(env_path: pathlib.Path, key: str, value: str) -> None: + """Append or update a key=value pair in a .env file.""" + lines: list[str] = [] + found = False + new_line = f"{key}={value}" + if env_path.exists(): + text = env_path.read_text() + lines = text.splitlines(keepends=True) + for i, line in enumerate(lines): + stripped = line.lstrip() + if stripped.startswith(f"{key}=") or stripped.startswith(f"{key} ="): + lines[i] = new_line + "\n" + found = True + break + if found: + env_path.write_text("".join(lines)) + else: + with open(env_path, "a") as f: + if lines and not lines[-1].endswith("\n"): + f.write("\n") + f.write(new_line + "\n") + + def _secrets_from_env( env_vars: dict[str, str], ) -> list[dict[str, str]]: @@ -817,6 +852,13 @@ def _deploy( if not deployment_id and not name: default_name = _normalize_image_name(pathlib.Path.cwd().name) name = click.prompt("Deployment name", default=default_name) + env_file = _resolve_env_file_path(config_json, config) + if env_file is not None and name: + _save_to_dotenv(env_file, _DEPLOYMENT_NAME_ENV, name) + click.secho( + f" Saved {_DEPLOYMENT_NAME_ENV}={name} to {env_file}", + fg="green", + ) secrets = _secrets_from_env(env_vars) diff --git a/libs/cli/tests/unit_tests/test_deploy_helpers.py b/libs/cli/tests/unit_tests/test_deploy_helpers.py index e8287f6a3..454d08b2e 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/libs/cli/tests/unit_tests/test_deploy_helpers.py @@ -10,6 +10,8 @@ from langgraph_cli.cli import ( _normalize_image_name, _normalize_image_tag, _parse_env_from_config, + _resolve_env_file_path, + _save_to_dotenv, ) @@ -132,3 +134,73 @@ class TestParseEnvFromConfig: assert result["GOOD"] == "value" # EMPTY= gives empty string, not None, so it should be present assert result["EMPTY"] == "" + + +class TestResolveEnvFilePath: + def test_inline_dict_returns_none(self, tmp_path): + config_path = tmp_path / "langgraph.json" + config_path.touch() + assert _resolve_env_file_path({"env": {"FOO": "bar"}}, config_path) is None + + def test_string_returns_resolved_path(self, tmp_path): + config_path = tmp_path / "langgraph.json" + config_path.touch() + result = _resolve_env_file_path({"env": "my.env"}, config_path) + assert result == (tmp_path / "my.env").resolve() + + def test_no_env_field_returns_cwd_dotenv(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + config_path = tmp_path / "langgraph.json" + config_path.touch() + result = _resolve_env_file_path({}, config_path) + assert result == tmp_path / ".env" + + def test_empty_dict_returns_cwd_dotenv(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + config_path = tmp_path / "langgraph.json" + config_path.touch() + result = _resolve_env_file_path({"env": {}}, config_path) + assert result == tmp_path / ".env" + + +class TestSaveToDotenv: + def test_creates_new_file(self, tmp_path): + env_file = tmp_path / ".env" + _save_to_dotenv(env_file, "MY_KEY", "my_value") + assert env_file.read_text() == "MY_KEY=my_value\n" + + def test_appends_to_existing_file(self, tmp_path): + env_file = tmp_path / ".env" + env_file.write_text("EXISTING=val\n") + _save_to_dotenv(env_file, "NEW_KEY", "new_val") + assert env_file.read_text() == "EXISTING=val\nNEW_KEY=new_val\n" + + def test_appends_newline_if_missing(self, tmp_path): + env_file = tmp_path / ".env" + env_file.write_text("EXISTING=val") + _save_to_dotenv(env_file, "NEW_KEY", "new_val") + assert env_file.read_text() == "EXISTING=val\nNEW_KEY=new_val\n" + + def test_updates_existing_key(self, tmp_path): + env_file = tmp_path / ".env" + env_file.write_text("KEY1=old\nKEY2=keep\n") + _save_to_dotenv(env_file, "KEY1", "new") + content = env_file.read_text() + assert "KEY1=new\n" in content + assert "KEY2=keep\n" in content + assert "KEY1=old" not in content + + def test_updates_key_with_spaces(self, tmp_path): + env_file = tmp_path / ".env" + env_file.write_text("KEY1 = old\nKEY2=keep\n") + _save_to_dotenv(env_file, "KEY1", "new") + content = env_file.read_text() + assert "KEY1=new\n" in content + assert "KEY1 = old" not in content + + def test_preserves_other_lines(self, tmp_path): + env_file = tmp_path / ".env" + env_file.write_text("# comment\nA=1\nB=2\n") + _save_to_dotenv(env_file, "C", "3") + content = env_file.read_text() + assert content == "# comment\nA=1\nB=2\nC=3\n"