From 876a702edd7f106380c21968bae18cfdcb4e4c96 Mon Sep 17 00:00:00 2001 From: wakinniranye31 Date: Sat, 18 Jul 2026 15:24:46 +0100 Subject: [PATCH] Support initializing PowerBiClient with a TokenProvider PowerBiClient and PowerBiAuth now accept a token_provider argument (a TokenProvider instance) instead of a raw access_token string. client_id/client_secret/redirect_uri/scope become optional and the whole MSAL/OAuth login() flow is skipped when a provider is supplied - there's nothing left to authenticate. Every request now goes through PowerBiAuth.get_token(), which either delegates to the external TokenProvider or (for the existing MSAL flow) validates/refreshes as before and returns the current token. PowerBiSession.build_headers() calls this once per request rather than reading a cached string, so a custom TokenProvider can rotate or refresh its token transparently - the library never assumes the token it saw last time is still valid. Includes a built-in StaticTokenProvider for the common case of already having a plain token string with no rotation logic needed. Same pattern used by the Azure SDK and Google Auth libraries. Addresses review feedback on this PR: - Replaced the raw access_token: str parameter with a TokenProvider interface (this commit). - New parameters are typed str | None = None / list[str] | None = None (project targets Python 3.10+). - PowerBiAuth.__init__ now raises the same ValueError guard as PowerBiClient.__init__ when constructed directly with neither a token_provider nor a full set of credentials. - samples/use_client_with_access_token.py no longer uses a JWT-shaped placeholder; replaced with an unambiguous 'REPLACE_WITH_YOUR_ACCESS_TOKEN' string. Rebased onto upstream's token-expiration handling (f6ca626) and access-token-validation-before-request (8cc7163) changes, which build_headers()/get_token() now compose with rather than duplicate. Tests: added TestPowerBiClientWithTokenProvider (including a RotatingTokenProvider case proving get_token() is called fresh each time, not cached) and TestPowerBiAuthWithTokenProvider covering the direct-construction guard. Full suite: 135 passed. --- CHANGELOG.md | 8 +++ powerbi/__init__.py | 4 ++ powerbi/auth.py | 86 +++++++++++++++++++------ powerbi/client.py | 53 +++++++++++---- powerbi/session.py | 5 +- powerbi/token_provider.py | 35 ++++++++++ samples/use_client_with_access_token.py | 30 +++++++++ tests/test_client.py | 80 +++++++++++++++++++++++ 8 files changed, 267 insertions(+), 34 deletions(-) create mode 100644 powerbi/token_provider.py create mode 100644 samples/use_client_with_access_token.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2de5451..67dd874 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (supports both My Workspace and In Group). - **reports**: `update_report_content` — updates report content from a source report (supports both My Workspace and In Group; replaces `update_report_content_in_group`). +- `TokenProvider` abstract base class and a built-in `StaticTokenProvider` + implementation (`powerbi.token_provider`), exported from `powerbi`. + `PowerBiClient` and `PowerBiAuth` now accept a `token_provider` argument, + letting callers skip the MSAL/OAuth login flow entirely when they already + have a token (e.g. from a service principal or managed identity) — every + request calls `token_provider.get_token()` fresh rather than reading a + stored string, so a custom provider can transparently rotate/refresh its + token. See `samples/use_client_with_access_token.py`. ### Fixed - **push_datasets**: `post_dataset` now sends JSON body (`json_payload=`) instead of diff --git a/powerbi/__init__.py b/powerbi/__init__.py index f8d0899..b94795a 100644 --- a/powerbi/__init__.py +++ b/powerbi/__init__.py @@ -3,6 +3,7 @@ from __future__ import annotations from powerbi.client import PowerBiClient +from powerbi.token_provider import StaticTokenProvider, TokenProvider from powerbi.enums import ( ColumnAggregationMethods, ColumnDataTypes, @@ -41,6 +42,9 @@ __all__ = [ # Client "PowerBiClient", + # Token providers + "TokenProvider", + "StaticTokenProvider", # Enums "ColumnAggregationMethods", "ColumnDataTypes", diff --git a/powerbi/auth.py b/powerbi/auth.py index 8d86c8f..cb5b4a9 100644 --- a/powerbi/auth.py +++ b/powerbi/auth.py @@ -11,6 +11,8 @@ import msal +from powerbi.token_provider import TokenProvider + logger = logging.getLogger(__name__) @@ -21,32 +23,37 @@ class PowerBiAuth: def __init__( self, - client_id: str, - client_secret: str, - redirect_uri: str, - scope: list[str], + client_id: str | None = None, + client_secret: str | None = None, + redirect_uri: str | None = None, + scope: list[str] | None = None, account_type: str = "common", credentials: str = None, + token_provider: TokenProvider | None = None, ): """Initializes the `PowerBiAuth` Client. ### Parameters ---- - client_id : str + client_id : str (optional, Default=None) The application Client ID assigned when - creating a new Microsoft App. + creating a new Microsoft App. Not required + if `token_provider` is provided. - client_secret : str + client_secret : str (optional, Default=None) The application Client Secret assigned when - creating a new Microsoft App. + creating a new Microsoft App. Not required + if `token_provider` is provided. - redirect_uri : str + redirect_uri : str (optional, Default=None) The application Redirect URI assigned when - creating a new Microsoft App. + creating a new Microsoft App. Not required + if `token_provider` is provided. - scope : List[str] + scope : list[str] (optional, Default=None) The list of scopes you want the application - to have access to. + to have access to. Not required if + `token_provider` is provided. account_type : str (optional, Default='common') The account type you're application wants to @@ -54,8 +61,23 @@ def __init__( credentials : str (optional, Default=None) The file path to your local credential file. + + token_provider : TokenProvider (optional, Default=None) + A `TokenProvider` (e.g. `StaticTokenProvider`) that supplies + an already-acquired access token. When provided, no MSAL app + is created and `login()` becomes a no-op, since there's + nothing left to authenticate — every request instead calls + `token_provider.get_token()` to get a current token. """ + if not token_provider and not all( + [client_id, client_secret, redirect_uri, scope] + ): + raise ValueError( + "Either `token_provider` or all of `client_id`, `client_secret`, " + "`redirect_uri`, and `scope` must be provided." + ) + self.credentials = credentials self.token_dict = None @@ -71,13 +93,37 @@ def __init__( self.refresh_token = None self._redirect_code = None + self._token_provider = token_provider - # Initialize the Credential App. - self.client_app = msal.ConfidentialClientApplication( - client_id=self.client_id, - authority=self.AUTHORITY_URL + self.account_type, - client_credential=self.client_secret, - ) + # Skip building the Credential App entirely when an external token + # provider was handed to us, since there's nothing left to authenticate. + if self._token_provider is not None: + self.client_app = None + else: + self.client_app = msal.ConfidentialClientApplication( + client_id=self.client_id, + authority=self.AUTHORITY_URL + self.account_type, + client_credential=self.client_secret, + ) + + def get_token(self) -> str: + """Return a current, valid access token. + + Delegates to the external `TokenProvider` if one was supplied; + otherwise validates/refreshes the MSAL-acquired token as needed + and returns it. + + ### Returns + ---- + str : + A valid access token. + """ + + if self._token_provider is not None: + return self._token_provider.get_token() + + self._token_validation() + return self.access_token def _load_or_save_credentials(self, action: str, token_dict: dict = None) -> bool: """Loads or saves the credential state for the Client Library. @@ -229,6 +275,10 @@ def _silent_sso(self) -> bool: def login(self) -> None: """Logs the user into the session.""" + # An external token provider was handed to us, nothing left to do. + if self._token_provider is not None: + return + # Load the State. self._load_or_save_credentials(action="load") diff --git a/powerbi/client.py b/powerbi/client.py index 75f2346..3f4a390 100644 --- a/powerbi/client.py +++ b/powerbi/client.py @@ -4,6 +4,7 @@ from powerbi.session import PowerBiSession from powerbi.auth import PowerBiAuth +from powerbi.token_provider import TokenProvider from powerbi.dashboards import Dashboards from powerbi.groups import Groups from powerbi.users import Users @@ -33,32 +34,37 @@ class PowerBiClient: def __init__( self, - client_id: str, - client_secret: str, - redirect_uri: str, - scope: list[str], + client_id: str | None = None, + client_secret: str | None = None, + redirect_uri: str | None = None, + scope: list[str] | None = None, account_type: str = "common", credentials: str = None, + token_provider: TokenProvider | None = None, ): """Initializes the Graph Client. ### Parameters ---- - client_id : str + client_id : str (optional, Default=None) The application Client ID assigned when - creating a new Microsoft App. + creating a new Microsoft App. Not required + if `token_provider` is provided. - client_secret : str + client_secret : str (optional, Default=None) The application Client Secret assigned when - creating a new Microsoft App. + creating a new Microsoft App. Not required + if `token_provider` is provided. - redirect_uri : str + redirect_uri : str (optional, Default=None) The application Redirect URI assigned when - creating a new Microsoft App. + creating a new Microsoft App. Not required + if `token_provider` is provided. - scope : List[str] + scope : list[str] (optional, Default=None) The list of scopes you want the application - to have access to. + to have access to. Not required if + `token_provider` is provided. account_type : str (optional, Default='common') The account type you're application wants to @@ -67,6 +73,14 @@ def __init__( credentials : str (optional, Default=None) The file path to your local credential file. + token_provider : TokenProvider (optional, Default=None) + A `TokenProvider` (e.g. `StaticTokenProvider`) supplying an + already-acquired Power BI access token (for example, obtained + through a service principal, managed identity, or an auth + flow handled outside this library). When provided, the + standard `client_id`/`client_secret`/`redirect_uri`/`scope` + login flow is skipped entirely. + ### Usage ---- >>> power_bi_client = PowerBiClient( @@ -76,8 +90,22 @@ def __init__( redirect_uri=redirect_uri, credentials='config/power_bi_state.jsonc' ) + + >>> # Or, with a token you already have: + >>> from powerbi import StaticTokenProvider + >>> power_bi_client = PowerBiClient( + token_provider=StaticTokenProvider(my_access_token) + ) """ + if not token_provider and not all( + [client_id, client_secret, redirect_uri, scope] + ): + raise ValueError( + "Either `token_provider` or all of `client_id`, `client_secret`, " + "`redirect_uri`, and `scope` must be provided." + ) + self.credentials = credentials self.client_id = client_id self.client_secret = client_secret @@ -92,6 +120,7 @@ def __init__( scope=self.scope, account_type=self.account_type, credentials=self.credentials, + token_provider=token_provider, ) self.power_bi_auth_client.login() diff --git a/powerbi/session.py b/powerbi/session.py index 88fadfd..52b5f6d 100644 --- a/powerbi/session.py +++ b/powerbi/session.py @@ -58,11 +58,8 @@ def build_headers(self) -> Dict: A dictionary containing all the components. """ - # Ensure the access token is still valid before making a request. - self.client._token_validation() - headers = { - "Authorization": f"Bearer {self.client.access_token}", + "Authorization": f"Bearer {self.client.get_token()}", "Content-Type": "application/json", } diff --git a/powerbi/token_provider.py b/powerbi/token_provider.py new file mode 100644 index 0000000..2526f08 --- /dev/null +++ b/powerbi/token_provider.py @@ -0,0 +1,35 @@ +"""Token provider abstraction for supplying Power BI access tokens.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class TokenProvider(ABC): + """Abstract base for anything that can supply a valid access token. + + Implementations are responsible for deciding internally whether their + token is still valid and fetching/refreshing it if not. Callers should + call `get_token()` immediately before each use rather than caching the + returned value themselves. + """ + + @abstractmethod + def get_token(self) -> str: + """Return a valid access token.""" + + +class StaticTokenProvider(TokenProvider): + """Wraps an already-acquired access token that this library will not + attempt to refresh or validate. + + Suitable when the caller manages token acquisition/refresh themselves + (e.g. a service principal client-credentials flow, a managed identity, + or any auth flow handled outside this library). + """ + + def __init__(self, token: str) -> None: + self._token = token + + def get_token(self) -> str: + return self._token diff --git a/samples/use_client_with_access_token.py b/samples/use_client_with_access_token.py new file mode 100644 index 0000000..bb45956 --- /dev/null +++ b/samples/use_client_with_access_token.py @@ -0,0 +1,30 @@ +"""Demonstrates initializing `PowerBiClient` with a token you already have. + +Useful when the access token was acquired elsewhere — e.g. a service +principal client-credentials flow, a managed identity, or another +part of your application — and you don't want this library to run +its own MSAL/OAuth login flow on top of that. +""" + +from powerbi.client import PowerBiClient +from powerbi.token_provider import StaticTokenProvider + +# Assume this came from your own auth flow (MSAL, a managed identity, +# a secrets manager, etc.) rather than this library. This placeholder is +# obviously not a real token — replace it with the token string your own +# auth flow produces. +existing_access_token = "REPLACE_WITH_YOUR_ACCESS_TOKEN" + +# Wrap it in a StaticTokenProvider: it always returns the same string, +# since this library has no way to refresh a token it didn't acquire. +# If you need automatic refresh, implement your own TokenProvider subclass +# whose get_token() fetches/refreshes internally. +token_provider = StaticTokenProvider(existing_access_token) + +# Initialize the Client directly with the provider. `client_id`, +# `client_secret`, `redirect_uri`, and `scope` aren't needed in this +# path since there's no login flow to run. +power_bi_client = PowerBiClient(token_provider=token_provider) + +# Use it exactly like a normally-authenticated client. +dashboard_service = power_bi_client.dashboards() diff --git a/tests/test_client.py b/tests/test_client.py index 8d25908..0c7193b 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -23,6 +23,7 @@ from powerbi.dataflows import Dataflows from powerbi.datasets import Datasets from powerbi.imports import Imports +from powerbi.token_provider import StaticTokenProvider class TestPowerBiSession(TestCase): @@ -144,5 +145,84 @@ def tearDown(self) -> None: del self.power_bi_client +class TestPowerBiClientWithTokenProvider(TestCase): + """Covers constructing a `PowerBiClient` from a `TokenProvider`, + bypassing the MSAL/OAuth login flow entirely.""" + + def test_skips_login_flow_and_delegates_to_provider(self): + """No `client_id`/`client_secret`/etc. and no MSAL app should + be needed when a `token_provider` is supplied, and `get_token()` + should delegate to it rather than reading a stored string.""" + + provider = StaticTokenProvider("my-existing-token") + power_bi_client = PowerBiClient(token_provider=provider) + + self.assertIsInstance(power_bi_client, PowerBiClient) + self.assertIsInstance(power_bi_client.power_bi_auth_client, PowerBiAuth) + self.assertIsNone(power_bi_client.power_bi_auth_client.client_app) + self.assertEqual( + power_bi_client.power_bi_auth_client.get_token(), "my-existing-token" + ) + + def test_login_is_a_no_op_when_token_provider_supplied(self): + """Calling `login()` again shouldn't attempt any MSAL/OAuth + work or change what `get_token()` returns.""" + + provider = StaticTokenProvider("my-existing-token") + power_bi_client = PowerBiClient(token_provider=provider) + power_bi_client.power_bi_auth_client.login() + + self.assertEqual( + power_bi_client.power_bi_auth_client.get_token(), "my-existing-token" + ) + + def test_get_token_reflects_provider_returning_a_new_value(self): + """`get_token()` should call the provider fresh each time, not + cache the first value it saw — this is what lets a custom + provider rotate/refresh its token transparently.""" + + class RotatingTokenProvider(StaticTokenProvider): + def __init__(self): + super().__init__("first-token") + self._calls = 0 + + def get_token(self) -> str: + self._calls += 1 + return f"token-{self._calls}" + + provider = RotatingTokenProvider() + power_bi_client = PowerBiClient(token_provider=provider) + + self.assertEqual(power_bi_client.power_bi_auth_client.get_token(), "token-1") + self.assertEqual(power_bi_client.power_bi_auth_client.get_token(), "token-2") + + def test_raises_without_token_provider_or_full_credentials(self): + """Should refuse to construct without either a `token_provider` + or a complete set of client credentials.""" + + with self.assertRaises(ValueError): + PowerBiClient(client_id="only-a-client-id") + + +class TestPowerBiAuthWithTokenProvider(TestCase): + """Covers the same `token_provider` guard directly on `PowerBiAuth`, + since it can be constructed without going through `PowerBiClient`.""" + + def test_raises_without_token_provider_or_full_credentials(self): + """Constructing `PowerBiAuth` directly with no `token_provider` + and incomplete credentials should fail fast with a clear error, + not surface a confusing failure later when a request is made.""" + + with self.assertRaises(ValueError): + PowerBiAuth(client_id="only-a-client-id") + + def test_accepts_token_provider_with_no_other_arguments(self): + provider = StaticTokenProvider("my-existing-token") + auth = PowerBiAuth(token_provider=provider) + + self.assertIsNone(auth.client_app) + self.assertEqual(auth.get_token(), "my-existing-token") + + if __name__ == "__main__": unittest.main()