initial proposal for client credentials support in python sdk

Author: zackverham

src/posit/connect/external/databricks.py

@@ -28,6 +28,46 @@ def auth_type(self) -> str:
     def __call__(self, *args, **kwargs) -> CredentialsProvider:
         raise NotImplementedError
 
+# TODO: Refactor common behavior across different cred providers.
+
+class PositContentCredentialsProvider:
+    def __init__(self, client: Client):
+        self._client = client
+
+    def __call__(self) -> Dict[str, str]:
+        credentials = self._client.oauth.get_content_credentials()
+        access_token = credentials.get("access_token")
+        if access_token is None:
+            raise ValueError("Missing value for field 'access_token' in credentials.")
+        return {"Authorization": f"Bearer {access_token}"}
+
+class PositContentCredentialsStrategy:
+    def __init__(
+        self,
+        local_strategy: CredentialsStrategy,
+        client: Optional[Client] = None,
+    ):
+        self._local_strategy = local_strategy
+        self._client = client
+
+    def sql_credentials_provider(self, *args, **kwargs):
+        return lambda: self.__call__(*args, **kwargs)
+
+    def auth_type(self) -> str:
+        if is_local():
+            return self._local_strategy.auth_type()
+        else:
+            return "posit-oauth-integration"
+
+    def __call__(self, *args, **kwargs) -> CredentialsProvider:
+        if is_local():
+            return self._local_strategy(*args, **kwargs)
+
+        if self._client is None:
+            self._client = Client()
+
+        return PositContentCredentialsProvider(self._client)
+
 
 class PositCredentialsProvider:
     def __init__(self, client: Client, user_session_token: str):

src/posit/connect/oauth/oauth.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from typing import Optional
 
 from typing_extensions import TypedDict
@@ -8,6 +9,27 @@
 from .integrations import Integrations
 from .sessions import Sessions
 
+GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
+USER_SESSION_TOKEN_TYPE = "urn:posit:connect:user-session-token"
+CONTENT_SESSION_TOKEN_TYPE = "urn:posit:connect:content-session-token"
+
+def _get_content_session_token() -> str:
+    """Return the content session token.
+
+    Reads the environment variable 'CONNECT_CONTENT_SESSION_TOKEN'.
+
+    Raises
+    ------
+        ValueError: If CONNECT_CONTENT_SESSION_TOKEN is not set or invalid
+
+    Returns
+    -------
+        str
+    """
+    value = os.environ.get("CONNECT_CONTENT_SESSION_TOKEN")
+    if not value:
+        raise ValueError("Invalid value for 'CONNECT_CONTENT_SESSION_TOKEN': Must be a non-empty string.")
+    return value
 
 class OAuth(Resources):
     def __init__(self, params: ResourceParameters, api_key: str) -> None:
@@ -27,14 +49,25 @@ def get_credentials(self, user_session_token: Optional[str] = None) -> Credentia
 
         # craft a credential exchange request
         data = {}
-        data["grant_type"] = "urn:ietf:params:oauth:grant-type:token-exchange"
-        data["subject_token_type"] = "urn:posit:connect:user-session-token"
+        data["grant_type"] = GRANT_TYPE
+        data["subject_token_type"] = USER_SESSION_TOKEN_TYPE
         if user_session_token:
             data["subject_token"] = user_session_token
 
         response = self.params.session.post(url, data=data)
         return Credentials(**response.json())
 
+    def get_content_credentials(self, content_session_token: Optional[str] = None) -> Credentials:
+        url = self.params.url + "v1/oauth/integrations/credentials"
+
+        # craft a credential exchange request
+        data = {}
+        data["grant_type"] = GRANT_TYPE
+        data["subject_token_type"] = CONTENT_SESSION_TOKEN_TYPE
+        data["subject_token"] = content_session_token or _get_content_session_token() 
+
+        response = self.params.session.post(url, data=data)
+        return Credentials(**response.json())
 
 class Credentials(TypedDict, total=False):
     access_token: str