Adding workspace/account logic for python

.gitattributes

@@ -4,6 +4,8 @@ sdk/go/api_esc_test.go linguist-generated=false
 
 sdk/python/**/* linguist-generated=true
 sdk/python/pulumi_esc_sdk/esc_client.py linguist-generated=false
+sdk/python/pulumi_esc_sdk/workspace.py linguist-generated=false
+sdk/python/pulumi_esc_sdk/workspace_models.py linguist-generated=false
 sdk/python/test/**/* linguist-generated=false
 sdk/python/pyproject.toml linguist-generated=false
 

CHANGELOG_PENDING.md

@@ -4,6 +4,8 @@
   - All SDK now automatically read in configuration environment variables
   - Go SDK also automatically picks up configuration from CLI Pulumi accounts
   [#76](https://github.com/pulumi/esc-sdk/pull/76)
+- Adds support for reading credentials from disk to Python SDK
+  [#81](https://github.com/pulumi/esc-sdk/pull/81)
 
 ### Bug Fixes
 

sdk/python/pulumi_esc_sdk/esc_client.py

@@ -12,6 +12,7 @@
 import yaml
 import os
 from urllib.parse import urlparse, urlunparse
+import pulumi_esc_sdk.workspace as workspace
 
 
 class EscClient:
@@ -441,6 +442,14 @@ def default_config(host=None,
         access_token = os.getenv("PULUMI_ACCESS_TOKEN")
     if not host:
         host = os.getenv("PULUMI_BACKEND_URL")
+
+    if not access_token or not host:
+        account, backend_url = workspace.get_current_account(False)
+        if not access_token:
+            access_token = account.accessToken if account else None
+        if not host:
+            host = backend_url
+
     return configuration.Configuration(
         host=host,
         access_token=access_token,

sdk/python/pulumi_esc_sdk/workspace.py

@@ -0,0 +1,93 @@
+# Copyright 2025, Pulumi Corporation.  All rights reserved.
+
+"""
+Pulumi workspace and account logic for python SDK.
+This is a partial port of ESC and Pulumi CLI code found in
+https://github.com/pulumi/esc/tree/main/cmd/esc/cli/workspace
+"""
+
+from dataclasses import dataclass, field
+import json
+import os
+import pathlib
+from typing import Optional
+
+from pulumi_esc_sdk.workspace_models import Account, Credentials
+
+def get_pulumi_home_dir() -> str:
+    """
+    Returns the path of the '.pulumi' folder where Pulumi puts its artifacts.
+    """
+    # Allow the folder we use to be overridden by an environment variable
+    dir_env = os.getenv("PULUMI_HOME")
+    if dir_env:
+        return dir_env
+
+    # Otherwise, use the current user's home dir + .pulumi
+    home_dir = pathlib.Path.home()
+
+    return str(home_dir.joinpath(".pulumi"))
+
+def get_pulumi_path(*elem: str) -> str:
+    """
+    Returns the path to a file or directory under the '.pulumi' folder.
+    It joins the path of the '.pulumi' folder with elements passed as arguments.
+    """
+    home_dir = get_pulumi_home_dir()
+    return str(pathlib.Path(home_dir).joinpath(*elem))
+
+def get_esc_bookkeeping_dir() -> str:
+    """
+    Returns the path of the '.esc' folder inside Pulumi home dir.
+    """
+    return get_pulumi_path(".esc")
+
+def get_path_to_creds_file(dir: str) -> str:
+    """
+    Returns the path to the esc credentials file on disk.
+    """
+    return str(pathlib.Path(dir).joinpath("credentials.json"))
+
+def get_esc_current_account_name() -> Optional[str]:
+    """
+    Returns the current account name from the ESC credentials file.
+    """
+    creds_file = get_path_to_creds_file(get_esc_bookkeeping_dir())
+    try:
+        with open(creds_file, 'r') as f:
+            data = json.loads(f.read())
+            return data['name']
+    except FileNotFoundError:
+        return None
+    except Exception as e:
+        print(f"An unexpected error occured: {e}")
+        return None
+
+def get_stored_credentials() -> Credentials:
+    """
+    Reads and parses credentials from the Pulumi credentials file.
+    """
+    creds_file = get_path_to_creds_file(get_pulumi_path())
+    try:
+        with open(creds_file, 'r') as f:
+            data = f.read()
+            creds = Credentials.from_json(data)
+            return creds
+    except FileNotFoundError:
+        return None
+    except Exception as e:
+        print(f"An unexpected error occured: {e}")
+        return None
+
+def get_current_account(shared) -> tuple[Account, str]:
+    """
+    Gets current account values from credentials file.
+    """
+    backend_url = get_esc_current_account_name()
+    pulumi_credentials = get_stored_credentials()
+    if not pulumi_credentials:
+        return None, None
+    if backend_url is None or shared:
+        backend_url = pulumi_credentials.current
+    pulumi_account = pulumi_credentials.accounts[backend_url]
+    return pulumi_account, backend_url

sdk/python/pulumi_esc_sdk/workspace_models.py

@@ -0,0 +1,45 @@
+# Copyright 2025, Pulumi Corporation.  All rights reserved.
+
+"""
+Models for Pulumi workspace and account logic for python SDK.
+This is a partial port of ESC and Pulumi CLI code found in
+https://github.com/pulumi/esc/tree/main/cmd/esc/cli/workspace
+"""
+
+from dataclasses import dataclass, field
+import json
+from typing import Dict, List, Optional
+
+@dataclass
+class Account:
+    accessToken: str
+    username: str
+    organizations: List[str]
+
+    @classmethod
+    def from_json(self, data: dict):
+        return self(
+            accessToken=data.get("accessToken"),
+            username=data.get("username"),
+            organizations=data.get("organizations", []),
+        )
+
+@dataclass
+class Credentials:
+    current: str
+    accessTokens: Dict[str, str]
+    accounts: Dict[str, Account]
+
+    @classmethod
+    def from_json(self, dataJson: str):
+        data = json.loads(dataJson)
+        accounts: Dict[str, Account] = {}
+        accounts_dict = data.get("accounts")
+        if accounts_dict:
+            for account_name, account_data in accounts_dict.items():
+                accounts[account_name] = Account.from_json(account_data)
+        return self(
+            current=data.get("current"),
+            accessTokens=data.get("accessTokens"),
+            accounts=accounts
+        )

sdk/python/test/test_esc_api.py

@@ -16,7 +16,7 @@ class TestEscApi(unittest.TestCase):
     """EscApi unit test stubs"""
 
     def setUp(self) -> None:
-        self.orgName = "pulumi"
+        self.orgName = os.getenv("PULUMI_ORG")
         self.assertIsNotNone(self.orgName, "PULUMI_ORG must be set")
 
         self.client = esc.esc_client.default_client()

sdk/python/test/test_pulumi_home/credentials.json

@@ -0,0 +1,25 @@
+{
+    "current": "https://api.moolumi.com",
+    "accessTokens": {
+        "https://api.moolumi.com": "pul-fake-token-moo",
+        "https://api.boolumi.com": "pul-fake-token-boo"
+    },
+    "accounts": {
+        "https://api.moolumi.com": {
+            "accessToken": "pul-fake-token-moo",
+            "username": "moolumipus",
+            "organizations": [
+                "moolumipus-org"
+            ],
+            "lastValidatedAt": "2025-03-17T11:19:28.392525488-06:00"
+        },
+        "https://api.boolumi.com": {
+            "accessToken": "pul-fake-token-boo",
+            "username": "boolumipus",
+            "organizations": [
+                "boolumipus-org2"
+            ],
+            "lastValidatedAt": "2026-03-17T11:19:28.392525488-06:00"
+        }
+    }
+}

sdk/python/test/test_pulumi_home_esc/.esc/credentials.json

@@ -0,0 +1,3 @@
+{
+    "name": "https://api.boolumi.com"
+}

sdk/python/test/test_pulumi_home_esc/credentials.json

@@ -0,0 +1,25 @@
+{
+    "current": "https://api.moolumi.com",
+    "accessTokens": {
+        "https://api.moolumi.com": "pul-fake-token-moo",
+        "https://api.boolumi.com": "pul-fake-token-boo"
+    },
+    "accounts": {
+        "https://api.moolumi.com": {
+            "accessToken": "pul-fake-token-moo",
+            "username": "moolumipus",
+            "organizations": [
+                "moolumipus-org"
+            ],
+            "lastValidatedAt": "2025-03-17T11:19:28.392525488-06:00"
+        },
+        "https://api.boolumi.com": {
+            "accessToken": "pul-fake-token-boo",
+            "username": "boolumipus",
+            "organizations": [
+                "boolumipus-org2"
+            ],
+            "lastValidatedAt": "2026-03-17T11:19:28.392525488-06:00"
+        }
+    }
+}

sdk/python/test/test_workspace_accounts.py

@@ -0,0 +1,51 @@
+# coding: utf-8
+
+# Copyright 2024, Pulumi Corporation.  All rights reserved.
+
+from typing import Optional
+import unittest
+import os
+
+import pulumi_esc_sdk as esc
+
+class TestWorkspaceAccounts(unittest.TestCase):
+    """WorkspaceAccounts unit test stubs"""
+    tokenBefore: Optional[str]
+    backendBefore: Optional[str]
+    homeBefore: Optional[str]
+
+    def setUp(self) -> None:
+        self.tokenBefore = os.getenv("PULUMI_ACCESS_TOKEN")
+        self.backendBefore = os.getenv("PULUMI_BACKEND_URL")
+        self.homeBefore = os.getenv("PULUMI_HOME")
+        os.environ["PULUMI_ACCESS_TOKEN"] = ""
+        os.environ["PULUMI_BACKEND_URL"] = ""
+
+    def tearDown(self) -> None:
+        os.environ["PULUMI_ACCESS_TOKEN"] = self.tokenBefore or ''
+        os.environ["PULUMI_BACKEND_URL"] = self.backendBefore or ''
+        os.environ["PULUMI_HOME"] = self.homeBefore or ''
+
+    def test_no_creds_at_all(self):
+        os.environ["PULUMI_HOME"] = "/not_real_dir"
+        self.client = esc.esc_client.default_client()
+        self.config = self.client.esc_api.api_client.configuration
+        self.assertEqual(self.config.host, "https://api.pulumi.com/api/esc")
+        self.assertTrue('Authorization' not in self.config.api_key)
+
+    def test_just_pulumi_creds(self):
+        os.environ["PULUMI_HOME"] = os.getcwd() + "/test/test_pulumi_home"
+        self.client = esc.esc_client.default_client()
+        self.config = self.client.esc_api.api_client.configuration
+        self.assertEqual(self.config.host, "https://api.moolumi.com/api/esc")
+        self.assertTrue(self.config.api_key['Authorization'], "pul-fake-token-moo")
+
+    def test_pulumi_creds_with_esc(self):
+        os.environ["PULUMI_HOME"] = os.getcwd() + "/test/test_pulumi_home_esc"
+        self.client = esc.esc_client.default_client()
+        self.config = self.client.esc_api.api_client.configuration
+        self.assertEqual(self.config.host, "https://api.boolumi.com/api/esc")
+        self.assertTrue(self.config.api_key['Authorization'], "pul-fake-token-boo")
+
+if __name__ == '__main__':
+    unittest.main()