Add async get_raw_access_token() method to AsyncPipedream for non-blocking OAuth token fetching

Co-authored-by: jverce <[email protected]>

Author: Copilot

src/pipedream/core/client_wrapper.py

@@ -120,3 +120,8 @@ async def async_get_headers(self) -> typing.Dict[str, str]:
             token = await self._async_token()
             headers["Authorization"] = f"Bearer {token}"
         return headers
+
+    async def _async_get_token(self) -> typing.Optional[str]:
+        if self._async_token is not None:
+            return await self._async_token()
+        return self._get_token()

src/pipedream/pipedream.py

@@ -110,10 +110,23 @@ def __init__(
     @property
     def raw_access_token(self) -> Optional[str]:
         """
-        Returns an access token that can be used to authenticate API requests
+        Returns an access token that can be used to authenticate API requests.
+
+        Note: When using OAuth authentication (client_id/client_secret), this property
+        may perform blocking network operations. For async applications, prefer using
+        the async method `get_raw_access_token()` instead.
         """
         return self._client_wrapper._get_token()
 
+    async def get_raw_access_token(self) -> Optional[str]:
+        """
+        Asynchronously returns an access token that can be used to authenticate API requests.
+
+        This method is non-blocking and safe to use in async contexts such as FastAPI,
+        Django ASGI, or any other asyncio-based application.
+        """
+        return await self._client_wrapper._async_get_token()
+
 
 def _get_base_url(environment: PipedreamEnvironment) -> str:
     """

tests/custom/test_client.py

@@ -1,7 +1,49 @@
+from unittest.mock import AsyncMock
+
 import pytest
 
+from pipedream import AsyncPipedream, Pipedream
+
 
 # Get started with writing tests with pytest at https://docs.pytest.org
 @pytest.mark.skip(reason="Unimplemented")
 def test_client() -> None:
     assert True
+
+
+def test_sync_pipedream_raw_access_token() -> None:
+    """Test synchronous Pipedream client raw_access_token property."""
+    client = Pipedream(access_token="test-token", project_id="test-project")
+    assert client.raw_access_token == "test-token"
+
+
+def test_async_pipedream_raw_access_token() -> None:
+    """Test AsyncPipedream client raw_access_token property with access_token."""
+    client = AsyncPipedream(access_token="test-token", project_id="test-project")
+    assert client.raw_access_token == "test-token"
+
+
+async def test_async_pipedream_get_raw_access_token() -> None:
+    """Test AsyncPipedream async method get_raw_access_token() with access_token."""
+    client = AsyncPipedream(access_token="test-token", project_id="test-project")
+    token = await client.get_raw_access_token()
+    assert token == "test-token"
+
+
+async def test_async_pipedream_get_raw_access_token_with_oauth() -> None:
+    """Test AsyncPipedream async method with OAuth flow."""
+    client = AsyncPipedream(
+        client_id="test-client-id",
+        client_secret="test-client-secret",
+        project_id="test-project",
+    )
+
+    # The client should have _async_token set when using OAuth
+    assert client._client_wrapper._async_token is not None
+
+    # Mock the async token provider
+    client._client_wrapper._async_token = AsyncMock(return_value="mocked-oauth-token")
+
+    # Test the async method
+    token = await client.get_raw_access_token()
+    assert token == "mocked-oauth-token"