Metadata-Version: 2.5
Name: dynamics-client-gh
Version: 0.1.6
Summary: Client library to easaly work within the ms dynamics universe
Author-email: PommereningMartin <m.pommerening@gruenhorn.de>, MichaelWehling <m.wehling@gruenhorn.de>
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.11
Requires-Dist: azure-identity>=1.26.0
Requires-Dist: azure-keyvault-secrets>=4.11.3
Requires-Dist: requests>=2.34.2
Description-Content-Type: text/markdown

# dynamics-client-gh

[![PyPI version](https://img.shields.io/pypi/v/dynamics-client-gh.svg)](https://pypi.org/project/dynamics-client-gh/)
![Python](https://img.shields.io/badge/python-3.11%2B-blue.svg)
![License](https://img.shields.io/badge/license-MIT-green.svg)

A small, typed Python client for the Microsoft Dataverse (Dynamics 365) Web API, with built-in Azure AD authentication by client secret or by certificate from Azure Key Vault.

```python
from dynamics_client import DataverseClient, TokenManager

token_manager = TokenManager(
    tenant_id="...",
    client_id="...",
    client_secret="...",
    resource="https://yourorg.crm.dynamics.com",
)
client = DataverseClient(token_manager=token_manager, base_url="https://yourorg.crm.dynamics.com")

accounts = client.get_list("accounts", query="$select=name&$top=5")
```

## Contents

- [Features](#features)
- [Installation](#installation)
- [Prerequisites](#prerequisites)
- [Authentication](#authentication)
- [Working with records](#working-with-records)
- [Error handling](#error-handling)
- [API reference](#api-reference)
- [Custom token managers](#custom-token-managers)
- [Development](#development)
- [License](#license)

## Features

- **Two authentication flows**: client secret (`TokenManager`) or certificate stored in Azure Key Vault (`CertificateTokenManager`).
- **Automatic token handling**: tokens are cached and refreshed 60 seconds before they expire.
- **Thread-safe**: one token manager can be shared across threads.
- **Simple CRUD API**: `get`, `get_list`, `create`, `update`, `delete` on any Dataverse entity set.
- **Typed**: ships with `py.typed` for editor completion and type checking.
- **Extensible**: add your own authentication flow by subclassing `BaseTokenManager`.

## Installation

```bash
pip install dynamics-client-gh
```

or with [uv](https://docs.astral.sh/uv/):

```bash
uv add dynamics-client-gh
```

Requires Python 3.11 or newer.

## Prerequisites

- An Azure AD (Entra ID) app registration with a client secret **or** a certificate.
- A Dataverse application user linked to that app registration, with a security role assigned.
- For certificate authentication: an Azure Key Vault that holds the certificate, and an identity that is allowed to read its secret (a managed identity in Azure, or your Azure CLI login locally).

## Authentication

### Client secret

```python
from dynamics_client import DataverseClient, TokenManager

token_manager = TokenManager(
    tenant_id="<tenant-id>",
    client_id="<app-client-id>",
    client_secret="<client-secret>",
    resource="https://yourorg.crm.dynamics.com",
)

client = DataverseClient(token_manager=token_manager, base_url="https://yourorg.crm.dynamics.com")
```

### Certificate from Azure Key Vault

The certificate is read from Key Vault at runtime, so no secret has to be stored in your application settings.

```python
from dynamics_client import DataverseClient

client = DataverseClient.with_certificate_auth(
    tenant_id="<tenant-id>",
    client_id="<app-client-id>",
    base_url="https://yourorg.crm.dynamics.com",
    key_vault_url="https://yourvault.vault.azure.net",
    certificate_secret_name="dataverse-cert",
)
```

Key Vault is accessed with [`DefaultAzureCredential`](https://learn.microsoft.com/python/api/azure-identity/azure.identity.defaultazurecredential): the managed identity when running in Azure, your Azure CLI login when running locally. Pass `managed_identity_client_id="..."` to select a specific user-assigned managed identity.

`with_certificate_auth` is a shortcut. To share one token manager between several clients, build it yourself:

```python
from dynamics_client import CertificateTokenManager, DataverseClient

token_manager = CertificateTokenManager(
    tenant_id="<tenant-id>",
    client_id="<app-client-id>",
    dataverse_url="https://yourorg.crm.dynamics.com",
    key_vault_url="https://yourvault.vault.azure.net",
    certificate_secret_name="dataverse-cert",
)

client = DataverseClient(token_manager=token_manager, base_url="https://yourorg.crm.dynamics.com")
```

## Working with records

Entity sets are addressed by their plural Web API name (`accounts`, `contacts`, ...). Queries use standard [OData query options](https://learn.microsoft.com/power-apps/developer/data-platform/webapi/query/overview).

```python
# List records
accounts = client.get_list("accounts", query="$select=name,revenue&$filter=revenue gt 100000&$top=10")

# Read one record
account = client.get("accounts", entity_id="00000000-0000-0000-0000-000000000001", query="$select=name")

# Create a record; returns the URL of the new record
record_url = client.create("accounts", {"name": "Contoso"})

# Update a record
client.update("accounts", "00000000-0000-0000-0000-000000000001", {"name": "Contoso Ltd"})

# Delete a record
client.delete("accounts", "00000000-0000-0000-0000-000000000001")
```

`get_list` returns the records of the first result page. For larger result sets, call `get` and follow `@odata.nextLink` in the response.

## Error handling

| Exception | Import from | Raised when |
| --- | --- | --- |
| `DataverseError` | `dynamics_client.dataverse_client` | a Dataverse API call fails (network error or HTTP error status) |
| `TokenAcquisitionError` | `dynamics_client` | the client-secret token request fails or returns an invalid response |
| `CertificateAuthError` | `dynamics_client.exceptions` | Key Vault access, certificate loading or the certificate token request fails |

```python
from dynamics_client import TokenAcquisitionError
from dynamics_client.dataverse_client import DataverseError
from dynamics_client.exceptions import CertificateAuthError

try:
    accounts = client.get_list("accounts")
except (TokenAcquisitionError, CertificateAuthError) as e:
    print(f"Authentication failed: {e}")
except DataverseError as e:
    print(f"Dataverse request failed: {e}")
    response = getattr(e.__cause__, "response", None)  # the underlying requests.Response, if any
    if response is not None:
        print(response.status_code, response.text)
```

## API reference

### `DataverseClient(token_manager, base_url, api_version="v9.2", timeout=30)`

| Method | Returns | Description |
| --- | --- | --- |
| `get(entity_set, entity_id=None, query="")` | `dict` | Read one record, or the raw collection response when `entity_id` is omitted. |
| `get_list(entity_set, query="")` | `list` | The `value` array of a collection response. |
| `create(entity_set, payload)` | `str` | Create a record and return its `OData-EntityId` URL. |
| `update(entity_set, entity_id, payload)` | `None` | Update an existing record (`PATCH` with `If-Match: *`, so it never creates one). |
| `delete(entity_set, entity_id)` | `None` | Delete a record. |
| `DataverseClient.with_certificate_auth(...)` | `DataverseClient` | Build a client that authenticates with a Key Vault certificate. |

`timeout` is the request timeout in seconds.

### `TokenManager(tenant_id, client_id, client_secret, resource, timeout=10)`

Client-credentials flow with a client secret. `resource` is the URL of your Dataverse environment.

### `CertificateTokenManager(tenant_id, client_id, dataverse_url, key_vault_url, certificate_secret_name, managed_identity_client_id=None)`

Client-credentials flow with a certificate. `certificate_secret_name` is the name of the certificate in Key Vault.

Both token managers provide:

| Method | Returns | Description |
| --- | --- | --- |
| `get_token()` | `str` | A valid access token, fetched or refreshed when needed. |
| `get_headers()` | `dict` | Request headers for the Dataverse Web API, including the `Authorization` header. |

## Custom token managers

To support another authentication flow, subclass `BaseTokenManager` and implement `_fetch_token()`. Caching, locking and `get_headers()` are inherited.

```python
import threading

from dynamics_client.base_token_manager import BaseTokenManager


class MyTokenManager(BaseTokenManager):
    def __init__(self):
        self._access_token = None
        self._expires_at = 0
        self._lock = threading.Lock()

    def _fetch_token(self):
        self._access_token = "..."   # the bearer token
        self._expires_at = 0         # expiry as a Unix timestamp
```

## Development

```bash
uv sync
uv run pytest src/tests
```

## License

MIT
