diff --git a/src/pipedream/pipedream.py b/src/pipedream/pipedream.py index 327b9ea..b1d86c9 100644 --- a/src/pipedream/pipedream.py +++ b/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: """ diff --git a/tests/custom/test_client.py b/tests/custom/test_client.py index ab04ce6..01c560a 100644 --- a/tests/custom/test_client.py +++ b/tests/custom/test_client.py @@ -1,7 +1,46 @@ +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", + ) + + # 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"