Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 4 additions & 0 deletions powerbi/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -41,6 +42,9 @@
__all__ = [
# Client
"PowerBiClient",
# Token providers
"TokenProvider",
"StaticTokenProvider",
# Enums
"ColumnAggregationMethods",
"ColumnDataTypes",
Expand Down
86 changes: 68 additions & 18 deletions powerbi/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@

import msal

from powerbi.token_provider import TokenProvider

logger = logging.getLogger(__name__)


Expand All @@ -21,41 +23,61 @@ 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
authenticate as.

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

Expand All @@ -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.
Expand Down Expand Up @@ -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")

Expand Down
53 changes: 41 additions & 12 deletions powerbi/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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(
Expand All @@ -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
Expand All @@ -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()
Expand Down
5 changes: 1 addition & 4 deletions powerbi/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
}

Expand Down
35 changes: 35 additions & 0 deletions powerbi/token_provider.py
Original file line number Diff line number Diff line change
@@ -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
30 changes: 30 additions & 0 deletions samples/use_client_with_access_token.py
Original file line number Diff line number Diff line change
@@ -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()
Loading