Add async get_raw_access_token() method to AsyncPipedream

src/pipedream/pipedream.py

@@ -110,10 +110,28 @@ 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 property `async_raw_access_token`
+        instead.
         """
         return self._client_wrapper._get_token()
 
+    @property
+    async def async_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.
+        """
+        if self._client_wrapper._async_token is not None:
+            return await self._client_wrapper._async_token()
+        return self.raw_access_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_async_raw_access_token() -> None:
+    """Test AsyncPipedream async method async_raw_access_token() with access_token."""
+    client = AsyncPipedream(access_token="test-token", project_id="test-project")
+    token = await client.async_raw_access_token
+    assert token == "test-token"
+
+
+async def test_async_pipedream_async_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.async_raw_access_token
+    assert token == "mocked-oauth-token"