Metadata-Version: 2.5
Name: dynamics-client-gh
Version: 0.1.7
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)
- [Working with files](#working-with-files)
- [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.
- **Automatic paging**: `get_list` follows `@odata.nextLink` and returns the records of all pages.
- **File columns**: `upload` and `download` stream files to and from Dataverse file columns, chunked for large files.
- **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` follows `@odata.nextLink` until the last page and returns the records of all pages in one list. The whole result is held in memory, so limit large tables with `$filter` or `$top`. To page manually, call `get` without an `entity_id`: it returns the raw collection response of one page, including `@odata.nextLink`.

## Working with files

`upload` and `download` work on [file columns](https://learn.microsoft.com/power-apps/developer/data-platform/file-attributes). `column` is the logical name of the file column.

```python
record_id = "00000000-0000-0000-0000-000000000001"

# Upload a local file to a file column
client.upload("accounts", record_id, "new_contract", "contract.pdf")

# Download the file column to disk; returns the destination as a Path
path = client.download("accounts", record_id, "new_contract", "downloads/contract.pdf")
```

- **Upload**: files up to 4 MB are sent in one request. Larger files are sent in chunks, with the chunk size set by the server. The file is read from disk as it is sent, not loaded into memory. The file name in Dataverse is the name of the local file.
- **Download**: the file is streamed to a temporary `.part` file next to the destination and renamed once it is complete, so a failed download never leaves a partial file under the destination name. Missing parent folders are created. If the size on disk does not match the size reported by the server, `DataverseError` is raised.

## Error handling

| Exception | Import from | Raised when |
| --- | --- | --- |
| `DataverseError` | `dynamics_client.dataverse_client` | a Dataverse API call fails (network error or HTTP error status), an upload session cannot be started, or a download is incomplete |
| `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 records of a collection, across all result pages. |
| `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. |
| `upload(entity_set, entity_id, column, file_path)` | `None` | Upload a local file to a file column, chunked when it is larger than 4 MB. |
| `download(entity_set, entity_id, column, dest)` | `Path` | Stream a file column to `dest` and return the path. |
| `DataverseClient.with_certificate_auth(...)` | `DataverseClient` | Build a client that authenticates with a Key Vault certificate. |

`timeout` is the request timeout in seconds. It applies to every single request, so each page of `get_list` and each chunk of `upload` gets the full timeout.

`create_upload_session`, `upload_chunk` and `verify_download` are the building blocks of `upload` and `download`. Call them directly only if you need to drive a chunked upload yourself.

### `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
```

The tests run in CI on every pull request and on pushes to `master`. Releases are described in [PUBLISHING.md](PUBLISHING.md).

## License

MIT
