feat(cli): warn about existing CLI tokens during datahub init --sso

After SSO login, query existing access tokens for the user and print a
count of CLI tokens with a link to manage them in the UI. This gives
users visibility into token accumulation without auto-revoking tokens
that may be in use elsewhere. Also removes the broken _revoke_old_cli_tokens
call that would crash at runtime and the unused List import.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 2 people

metadata-ingestion/src/datahub/cli/sso_cli.py

@@ -1,10 +1,15 @@
+import logging
 import urllib.parse
 from datetime import datetime
 from typing import Tuple
 
 import click
 import requests
 
+logger = logging.getLogger(__name__)
+
+CLI_TOKEN_PREFIX = "cli token "
+
 _INSTALL_HELP = """\
 The --sso flag requires Playwright and a Chromium browser.
 
@@ -33,6 +38,51 @@ def _check_playwright_ready() -> None:
         ) from e
 
 
+def _warn_about_existing_cli_tokens(
+    session: requests.Session,
+    frontend_url: str,
+    actor_urn: str,
+) -> None:
+    """Best-effort warning about existing CLI tokens for the current user."""
+    try:
+        response = session.post(
+            f"{frontend_url}/api/v2/graphql",
+            json={
+                "query": """query listAccessTokens($input: ListAccessTokenInput!) {
+                    listAccessTokens(input: $input) {
+                        total
+                        tokens { name }
+                    }
+                }""",
+                "variables": {
+                    "input": {
+                        "start": 0,
+                        "count": 100,
+                        "filters": [
+                            {
+                                "field": "ownerUrn",
+                                "values": [actor_urn],
+                            }
+                        ],
+                    }
+                },
+            },
+        )
+        response.raise_for_status()
+        data = response.json()
+        tokens = data.get("data", {}).get("listAccessTokens", {}).get("tokens", [])
+        cli_token_count = sum(
+            1 for t in tokens if t.get("name", "").startswith(CLI_TOKEN_PREFIX)
+        )
+        if cli_token_count > 0:
+            click.echo(
+                f"⚠ You have {cli_token_count} existing CLI token(s). "
+                f"Manage them at {frontend_url}/settings/tokens"
+            )
+    except Exception:
+        logger.debug("Failed to check existing CLI tokens", exc_info=True)
+
+
 def browser_sso_login(
     frontend_url: str,
     token_duration: str,
@@ -114,6 +164,8 @@ def browser_sso_login(
 
     click.echo(f"✓ Logged in as {actor_urn}")
 
+    _warn_about_existing_cli_tokens(session, frontend_url, actor_urn)
+
     # Generate an access token via the frontend GraphQL API
     now = datetime.now()
     timestamp = now.astimezone().isoformat()

metadata-ingestion/tests/unit/cli/test_sso_cli.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from datahub.cli.sso_cli import browser_sso_login
+from datahub.cli.sso_cli import _warn_about_existing_cli_tokens, browser_sso_login
 
 
 @pytest.fixture
@@ -64,8 +64,14 @@ def test_extracts_cookies_and_generates_token(self, mock_playwright: dict) -> No
         with patch("datahub.cli.sso_cli.requests") as mock_requests:
             mock_session = MagicMock()
             mock_requests.Session.return_value = mock_session
-            mock_response = MagicMock()
-            mock_response.json.return_value = {
+
+            # First call: listAccessTokens (warning check), second: createAccessToken
+            list_response = MagicMock()
+            list_response.json.return_value = {
+                "data": {"listAccessTokens": {"total": 0, "tokens": []}}
+            }
+            create_response = MagicMock()
+            create_response.json.return_value = {
                 "data": {
                     "createAccessToken": {
                         "accessToken": "generated-sso-token-xyz",
@@ -76,7 +82,7 @@ def test_extracts_cookies_and_generates_token(self, mock_playwright: dict) -> No
                     }
                 }
             }
-            mock_session.post.return_value = mock_response
+            mock_session.post.side_effect = [list_response, create_response]
 
             token_name, access_token = browser_sso_login(
                 "http://localhost:9002", "ONE_HOUR"
@@ -88,16 +94,16 @@ def test_extracts_cookies_and_generates_token(self, mock_playwright: dict) -> No
         # Verify cookies were set on the session
         assert mock_session.cookies.set.call_count == 2
 
-        # Verify GraphQL call was made
-        mock_session.post.assert_called_once()
-        call_args = mock_session.post.call_args
-        assert call_args[0][0] == "http://localhost:9002/api/v2/graphql"
-        assert "createAccessToken" in call_args[1]["json"]["query"]
+        # Verify GraphQL calls were made (list + create)
+        assert mock_session.post.call_count == 2
+        create_call = mock_session.post.call_args_list[1]
+        assert create_call[0][0] == "http://localhost:9002/api/v2/graphql"
+        assert "createAccessToken" in create_call[1]["json"]["query"]
         assert (
-            call_args[1]["json"]["variables"]["input"]["actorUrn"]
+            create_call[1]["json"]["variables"]["input"]["actorUrn"]
             == "urn:li:corpuser:john.doe"
         )
-        assert call_args[1]["json"]["variables"]["input"]["duration"] == "ONE_HOUR"
+        assert create_call[1]["json"]["variables"]["input"]["duration"] == "ONE_HOUR"
 
     def test_timeout_raises_error(self, mock_playwright: dict) -> None:
         """Verify timeout if login never completes."""
@@ -150,3 +156,42 @@ def test_graphql_error_raises(self, mock_playwright: dict) -> None:
 
             with pytest.raises(Exception, match="Failed to generate access token"):
                 browser_sso_login("http://localhost:9002", "ONE_HOUR")
+
+
+class TestWarnAboutExistingCliTokens:
+    def test_warns_about_existing_cli_tokens(
+        self, capsys: pytest.CaptureFixture
+    ) -> None:
+        session = MagicMock()
+        response = MagicMock()
+        response.json.return_value = {
+            "data": {
+                "listAccessTokens": {
+                    "total": 4,
+                    "tokens": [
+                        {"name": "cli token 2026-03-01T10:00:00"},
+                        {"name": "cli token 2026-03-02T10:00:00"},
+                        {"name": "cli token 2026-03-03T10:00:00"},
+                        {"name": "manually created token"},
+                    ],
+                }
+            }
+        }
+        session.post.return_value = response
+
+        _warn_about_existing_cli_tokens(
+            session, "https://example.com", "urn:li:corpuser:alice"
+        )
+
+        captured = capsys.readouterr()
+        assert "3 existing CLI token(s)" in captured.out
+        assert "https://example.com/settings/tokens" in captured.out
+
+    def test_warning_failure_does_not_block(self) -> None:
+        session = MagicMock()
+        session.post.side_effect = Exception("network error")
+
+        # Should not raise — failure is silently logged
+        _warn_about_existing_cli_tokens(
+            session, "https://example.com", "urn:li:corpuser:alice"
+        )