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
14 changes: 14 additions & 0 deletions .claude/skills/dataverse-sdk-use/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,20 @@ with DataverseClient("https://yourorg.crm.dynamics.com", credential) as client:
client = DataverseClient("https://yourorg.crm.dynamics.com", credential)
```

### Acquiring Tokens for Other Microsoft Resources

`client.auth.acquire_token(resource_url)` returns an OAuth2 token from the same credential for any AAD-protected Microsoft resource (e.g. a linked Finance & Operations environment). The `/.default` scope is appended automatically.

```python
# Token for a linked Finance & Operations environment
fno_token = client.auth.acquire_token("https://myenv.operations.dynamics.com")

# Use the F&O token to call F&O OData / Custom Service endpoints directly
headers = {"Authorization": f"Bearer {fno_token}"}
```

The customer's AAD app must already have the required permission on the target resource. For F&O the standard delegated permissions are `Odata.FullAccess` and `CustomService.FullAccess` on the **Microsoft Dynamics ERP** API (`00000015-0000-0000-c000-000000000000`).

### CRUD Operations

#### Create Records
Expand Down
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Added
- `client.auth.acquire_token(resource_url)` -- acquire an OAuth2 access token for any Microsoft AAD-protected resource (for example a linked Finance & Operations environment) using the same credential the Dataverse client was constructed with. The `/.default` scope is appended automatically; token caching and refresh remain the credential's responsibility. The internal Dataverse request path now goes through the same method, removing the inline scope construction in `_ODataClient._headers()`.

## [0.1.0b10] - 2026-05-12

### Added
Expand Down
13 changes: 13 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,19 @@ client = DataverseClient("https://yourorg.crm.dynamics.com", credential)

> **Complete authentication setup**: See **[Use OAuth with Dataverse](https://learn.microsoft.com/power-apps/developer/data-platform/authenticate-oauth)** for app registration, all credential types, and security configuration.

#### Acquiring tokens for other Microsoft resources

The same credential can be used to acquire tokens for any Microsoft AAD-protected resource the caller has access to -- notably a Finance & Operations environment linked to the same Dataverse org. Use `client.auth.acquire_token(resource_url)` to obtain a token without constructing a second credential:

```python
# Token for a linked Finance & Operations environment
fno_token = client.auth.acquire_token("https://myenv.operations.dynamics.com")

headers = {"Authorization": f"Bearer {fno_token}"}
```

The `/.default` scope is appended automatically. The customer's AAD app must already have the required permission on the target resource and admin consent granted. For Finance & Operations the standard permissions are `Odata.FullAccess` and `CustomService.FullAccess` on the **Microsoft Dynamics ERP** API (`00000015-0000-0000-c000-000000000000`).

## Key concepts

The SDK provides a simple, pythonic interface for Dataverse operations:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,20 @@ with DataverseClient("https://yourorg.crm.dynamics.com", credential) as client:
client = DataverseClient("https://yourorg.crm.dynamics.com", credential)
```

### Acquiring Tokens for Other Microsoft Resources

`client.auth.acquire_token(resource_url)` returns an OAuth2 token from the same credential for any AAD-protected Microsoft resource (e.g. a linked Finance & Operations environment). The `/.default` scope is appended automatically.

```python
# Token for a linked Finance & Operations environment
fno_token = client.auth.acquire_token("https://myenv.operations.dynamics.com")

# Use the F&O token to call F&O OData / Custom Service endpoints directly
headers = {"Authorization": f"Bearer {fno_token}"}
```

The customer's AAD app must already have the required permission on the target resource. For F&O the standard delegated permissions are `Odata.FullAccess` and `CustomService.FullAccess` on the **Microsoft Dynamics ERP** API (`00000015-0000-0000-c000-000000000000`).

### CRUD Operations

#### Create Records
Expand Down
57 changes: 53 additions & 4 deletions src/PowerPlatform/Dataverse/core/_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,12 @@
# Licensed under the MIT license.

"""
Authentication helpers for Dataverse.
Authentication helpers.

This module provides :class:`~PowerPlatform.Dataverse.core._auth._AuthManager`, a thin wrapper over any Azure Identity
``TokenCredential`` for acquiring OAuth2 access tokens, and :class:`~PowerPlatform.Dataverse.core._auth._TokenPair` for
storing the acquired token alongside its scope.
``TokenCredential`` for acquiring OAuth2 access tokens for Microsoft AAD-protected resources -- Dataverse by default,
and any other resource (e.g. a linked Finance & Operations environment) when an explicit scope is supplied --
and :class:`~PowerPlatform.Dataverse.core._auth._TokenPair` for storing the acquired token alongside its scope.
"""

from __future__ import annotations
Expand All @@ -33,7 +34,15 @@ class _TokenPair:

class _AuthManager:
"""
Azure Identity-based authentication manager for Dataverse.
Azure Identity-based authentication manager.

Resource-agnostic: the scope passed to :meth:`_acquire_token` selects
the target resource. The Dataverse client supplies its own
``<base_url>/.default`` scope on every internal request via
:meth:`acquire_token`, and the same method can be called externally
(through ``client.auth.acquire_token(...)``) to obtain tokens for
other Microsoft AAD-protected resources -- for example a linked
Finance & Operations environment.

:param credential: Azure Identity credential implementation.
:type credential: ~azure.core.credentials.TokenCredential
Expand All @@ -57,3 +66,43 @@ def _acquire_token(self, scope: str) -> _TokenPair:
"""
token = self.credential.get_token(scope)
return _TokenPair(resource=scope, access_token=token.token)

def acquire_token(self, resource_url: str) -> str:
"""
Acquire an OAuth2 access token for a Microsoft AAD-protected resource.

Resource-agnostic helper: pass the resource URL (Dataverse env URL
for Dataverse, Finance & Operations env URL for F&O, etc.) and the
``/.default`` scope suffix is appended automatically before
delegating to the underlying credential. Token caching, refresh,
and silent reauthentication are the credential's responsibility;
Azure Identity credentials cache in-memory by default so repeated
calls are cheap.

:param resource_url: Resource URL for the target Microsoft service
(for example ``"https://myenv.operations.dynamics.com"``).
Trailing slash is removed before scope construction.
:type resource_url: :class:`str`

:return: OAuth2 access token string suitable for placing in an
``Authorization: Bearer ...`` header.
:rtype: :class:`str`

:raises ValueError: If ``resource_url`` is empty after trimming.
:raises ~azure.core.exceptions.ClientAuthenticationError: If token
acquisition fails.

Example:
Acquire a token for a linked Finance & Operations environment
using the same credential the Dataverse client was built with::

client = DataverseClient(dv_url, credential)
fno_token = client.auth.acquire_token(
"https://myenv.operations.dynamics.com"
)
"""
target = (resource_url or "").rstrip("/")
if not target:
raise ValueError("resource_url must not be empty.")
scope = f"{target}/.default"
return self._acquire_token(scope).access_token
3 changes: 1 addition & 2 deletions src/PowerPlatform/Dataverse/data/_odata.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,8 +241,7 @@ def close(self) -> None:

def _headers(self) -> Dict[str, str]:
"""Build standard OData headers with bearer auth."""
scope = f"{self.base_url}/.default"
token = self.auth._acquire_token(scope).access_token
token = self.auth.acquire_token(self.base_url)
ua = _USER_AGENT
if self._operation_context:
ua = f"{_USER_AGENT} ({self._operation_context})"
Expand Down
10 changes: 9 additions & 1 deletion tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,12 @@

@pytest.fixture
def dummy_auth():
"""Mock authentication object for testing."""
"""Mock authentication object for testing.

Mirrors the real ``_AuthManager`` surface: both the internal
``_acquire_token(scope)`` (used directly by older tests) and the public
``acquire_token(resource_url)`` (used by ``_ODataClient._headers``).
"""

class DummyAuth:
def _acquire_token(self, scope):
Expand All @@ -24,6 +29,9 @@ class Token:

return Token()

def acquire_token(self, resource_url):
return self._acquire_token(f"{(resource_url or '').rstrip('/')}/.default").access_token

return DummyAuth()


Expand Down
43 changes: 43 additions & 0 deletions tests/unit/core/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,3 +33,46 @@ def test_acquire_token_returns_token_pair(self):
self.assertIsInstance(result, _TokenPair)
self.assertEqual(result.resource, "https://org.crm.dynamics.com/.default")
self.assertEqual(result.access_token, "my-access-token")

def test_acquire_token_public_appends_default_scope(self):
"""acquire_token appends /.default to the resource URL and returns the access_token string."""
mock_credential = MagicMock(spec=TokenCredential)
mock_credential.get_token.return_value = MagicMock(token="dv-token")

manager = _AuthManager(mock_credential)
result = manager.acquire_token("https://org.crm.dynamics.com")

mock_credential.get_token.assert_called_once_with("https://org.crm.dynamics.com/.default")
self.assertEqual(result, "dv-token")

def test_acquire_token_public_strips_trailing_slash(self):
"""acquire_token strips a trailing slash before constructing the scope."""
mock_credential = MagicMock(spec=TokenCredential)
mock_credential.get_token.return_value = MagicMock(token="t")

manager = _AuthManager(mock_credential)
manager.acquire_token("https://myenv.operations.dynamics.com/")

mock_credential.get_token.assert_called_once_with("https://myenv.operations.dynamics.com/.default")

def test_acquire_token_public_supports_alternate_resource(self):
"""acquire_token works for any resource URL (e.g. linked Finance & Operations env)."""
mock_credential = MagicMock(spec=TokenCredential)
mock_credential.get_token.return_value = MagicMock(token="fno-token")

manager = _AuthManager(mock_credential)
result = manager.acquire_token("https://myenv.operations.dynamics.com")

mock_credential.get_token.assert_called_once_with("https://myenv.operations.dynamics.com/.default")
self.assertEqual(result, "fno-token")

def test_acquire_token_public_empty_url_raises(self):
"""acquire_token raises ValueError when resource_url is empty after trim and does not call get_token."""
mock_credential = MagicMock(spec=TokenCredential)
manager = _AuthManager(mock_credential)

with self.assertRaises(ValueError):
manager.acquire_token("")
with self.assertRaises(ValueError):
manager.acquire_token("/")
mock_credential.get_token.assert_not_called()
3 changes: 3 additions & 0 deletions tests/unit/core/test_http_errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ class T:

return T()

def acquire_token(self, resource_url):
return self._acquire_token(f"{(resource_url or '').rstrip('/')}/.default").access_token


class DummyHTTP:
def __init__(self, responses):
Expand Down
3 changes: 3 additions & 0 deletions tests/unit/data/test_enum_optionset_payload.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@ class T:

return T()

def acquire_token(self, resource_url): # pragma: no cover - simple stub
return self._acquire_token(f"{(resource_url or '').rstrip('/')}/.default").access_token


class DummyConfig:
"""Minimal config stub providing attributes _ODataClient.__init__ expects."""
Expand Down
3 changes: 3 additions & 0 deletions tests/unit/data/test_logical_crud.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@ class T:

return T()

def acquire_token(self, resource_url):
return self._acquire_token(f"{(resource_url or '').rstrip('/')}/.default").access_token


class DummyHTTPClient:
def __init__(self, responses):
Expand Down
Loading