Last updated: 2026-04-06

Reference

Python Code Samples¶

Production-ready Python examples for integrating with Dxtra's privacy compliance API.

Prerequisites¶

Requirements

Python 3.11+ installed on your machine
A Dxtra API key - see Authentication Guide
Familiarity with Python async/await and type hints

Making a GraphQL Query¶

This example shows how to make authenticated requests to the Dxtra GraphQL API using the requests library.

First, install requests:

Bash

pip install requests

Then, use the following code to fetch a list of data controllers:

Python

import requests
import os

DXTRA_API_KEY = "YOUR_API_KEY"
DXTRA_AUTH_URL = "https://auth.dxtra.ai/v1/signin/pat"
DXTRA_API_URL = "https://api.dxtra.ai/v1/graphql"

def exchange_api_key_for_jwt(api_key):
    """Exchange API key for JWT token using PAT flow."""
    try:
        response = requests.post(DXTRA_AUTH_URL, json={
            "personalAccessToken": api_key
        })
        response.raise_for_status()
        return response.json()["session"]["accessToken"]
    except requests.exceptions.RequestException as e:
        print(f"Error exchanging API key for JWT: {e}")
        raise

def get_data_controllers():
    try:
        # Step 1: Exchange API key for JWT
        jwt_token = exchange_api_key_for_jwt(DXTRA_API_KEY)

        # Step 2: Use JWT to make GraphQL request
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {jwt_token}",
            "X-Hasura-Role": "user",
        }
        query = """
        query GetDataControllers {
          dataControllers {
            id
            title
            did
            createdAt
          }
        }
        """
        response = requests.post(
            DXTRA_API_URL, headers=headers, json={"query": query}
        )
        response.raise_for_status()
        data = response.json()

        # Handle GraphQL errors
        if "errors" in data:
            print(f"GraphQL Error: {data['errors'][0]['message']}")
            return None

        print(data["data"]["dataControllers"])
        return data["data"]["dataControllers"]
    except requests.exceptions.RequestException as e:
        print(f"Error fetching data controllers: {e}")
        return None

if __name__ == "__main__":
    get_data_controllers()

API Key Security

Replace YOUR_API_KEY with your actual Dxtra API key. Never commit API keys to version control - use environment variables instead.

This example demonstrates the proper two-step authentication flow:

Exchange API key for JWT: Use the PAT (Personal Access Token) endpoint
Make GraphQL requests: Use the JWT token with proper role headers

Advanced Examples¶

Environment Variables Setup¶

Secure Configuration Pattern

import os
from dataclasses import dataclass

@dataclass
class Config:
    """Dxtra API configuration from environment variables."""
    api_key: str
    auth_url: str = "https://auth.dxtra.ai/v1/signin/pat"
    api_url: str = "https://api.dxtra.ai/v1/graphql"

    @classmethod
    def from_env(cls):
        """Load configuration from environment variables."""
        api_key = os.getenv("DXTRA_API_KEY")
        if not api_key:
            raise ValueError("DXTRA_API_KEY environment variable is required")

        return cls(
            api_key=api_key,
            auth_url=os.getenv("DXTRA_AUTH_URL", cls.auth_url),
            api_url=os.getenv("DXTRA_API_URL", cls.api_url)
        )

config = Config.from_env()

Production-Ready Client Class¶

Production API Client

import requests
import time
from typing import Optional, Dict, Any
from datetime import datetime, timedelta

class DxtraApiClient:
    """Production-ready Dxtra API client with error handling and retry logic."""

    def __init__(self, api_key: str, auth_url: str = None, api_url: str = None):
        self.api_key = api_key
        self.auth_url = auth_url or "https://auth.dxtra.ai/v1/signin/pat"
        self.api_url = api_url or "https://api.dxtra.ai/v1/graphql"
        self.jwt_token: Optional[str] = None
        self.token_expires: Optional[datetime] = None

    def authenticate(self) -> str:
        """Exchange API key for JWT token."""
        try:
            response = requests.post(self.auth_url, json={
                "personalAccessToken": self.api_key
            })
            response.raise_for_status()

            session_data = response.json()["session"]
            self.jwt_token = session_data["accessToken"]

            # JWT access tokens expire in 15 minutes (900 seconds)
            # Refresh 2 minutes early to avoid expiration during requests
            expires_in = session_data.get("accessTokenExpiresIn", 900)
            self.token_expires = datetime.now() + timedelta(seconds=expires_in - 120)
            return self.jwt_token

        except requests.exceptions.RequestException as e:
            raise Exception(f"Authentication failed: {e}")

    def _ensure_valid_token(self) -> str:
        """Ensure we have a valid JWT token."""
        if not self.jwt_token or (self.token_expires and datetime.now() >= self.token_expires):
            return self.authenticate()
        return self.jwt_token

    def query(self, graphql_query: str, variables: Dict[str, Any] = None, retries: int = 3) -> Dict[str, Any]:
        """Execute a GraphQL query with error handling and retries."""
        variables = variables or {}
        self._ensure_valid_token()

        for attempt in range(1, retries + 1):
            try:
                headers = {
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {self.jwt_token}",
                    "X-Hasura-Role": "user",
                }

                response = requests.post(
                    self.api_url,
                    headers=headers,
                    json={"query": graphql_query, "variables": variables}
                )

                # Handle GraphQL errors
                if response.status_code == 200:
                    data = response.json()
                    if "errors" in data:
                        raise Exception(f"GraphQL Error: {data['errors'][0]['message']}")
                    return data["data"]

                response.raise_for_status()

            except requests.exceptions.RequestException as e:
                # Handle 401 by re-authenticating
                if hasattr(e, 'response') and e.response.status_code == 401 and attempt < retries:
                    print("Token expired, re-authenticating...")
                    self.authenticate()
                    continue

                # Handle rate limits with exponential backoff
                if hasattr(e, 'response') and e.response.status_code == 429 and attempt < retries:
                    delay = 2 ** attempt
                    print(f"Rate limited, retrying in {delay}s...")
                    time.sleep(delay)
                    continue

                if attempt == retries:
                    raise

                # Generic retry for other errors
                delay = 2 ** (attempt - 1)
                print(f"Request failed, retrying in {delay}s... (attempt {attempt}/{retries})")
                time.sleep(delay)

Type-Safe Async Client with GQL¶

For production applications, use the gql library for type-safe async GraphQL operations:

Type-Safe Async GraphQL Client

import asyncio
import os
from typing import Optional, Dict, Any, List
from dataclasses import dataclass

import aiohttp
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
from gql.transport.exceptions import TransportError

@dataclass
class DxtraAsyncClient:
    """Type-safe async Dxtra API client using gql library."""

    api_key: str
    auth_url: str = "https://auth.dxtra.ai/v1/signin/pat"
    api_url: str = "https://api.dxtra.ai/v1/graphql"
    jwt_token: Optional[str] = None
    _client: Optional[Client] = None
    _session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        """Context manager entry - authenticate and setup client."""
        self._session = aiohttp.ClientSession()
        await self.authenticate()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit - cleanup resources."""
        if self._client:
            await self._client.close_async()
        if self._session:
            await self._session.close()

    async def authenticate(self) -> str:
        """Exchange API key for JWT token."""
        async with self._session.post(
            self.auth_url,
            json={"personalAccessToken": self.api_key}
        ) as response:
            response.raise_for_status()
            data = await response.json()
            self.jwt_token = data["session"]["accessToken"]

            # Setup GraphQL client with authentication
            headers = {
                "Authorization": f"Bearer {self.jwt_token}",
                "X-Hasura-Role": "user"
            }
            transport = AIOHTTPTransport(url=self.api_url, headers=headers)
            self._client = Client(transport=transport, fetch_schema_from_transport=False)

            return self.jwt_token

    async def query(self, graphql_query: str, variables: Dict[str, Any] = None) -> Dict[str, Any]:
        """Execute async GraphQL query with type safety."""
        if not self._client:
            raise RuntimeError("Client not initialized. Use async context manager.")

        try:
            query_node = gql(graphql_query)
            result = await self._client.execute_async(
                query_node,
                variable_values=variables or {}
            )
            return result

        except TransportError as e:
            raise Exception(f"GraphQL transport error: {e}")

# Usage example
async def fetch_data_controllers():
    """Fetch data controllers using async client."""
    async with DxtraAsyncClient(os.getenv("DXTRA_API_KEY")) as client:
        query = """
        query GetDataControllers {
          dataControllers {
            id
            title
            did
            createdAt
          }
        }
        """

        data = await client.query(query)
        return data["dataControllers"]

# Run async function
if __name__ == "__main__":
    controllers = asyncio.run(fetch_data_controllers())
    print(f"Found {len(controllers)} data controllers")

Installation Requirements

For async support with type-safe GraphQL, install:

Bash

pip install gql[aiohttp] aiohttp python-dotenv

Using the Client¶

Client Usage Examples

import os
import asyncio
from dotenv import load_dotenv

load_dotenv()

async def main():
    """Example usage of Dxtra API clients."""

    # Basic sync client
    from requests_client import DxtraApiClient
    client = DxtraApiClient(os.getenv("DXTRA_API_KEY"))

    # Fetch data controllers
    query = """
    query GetDataControllers {
      dataControllers {
        id
        title
        did
        createdAt
      }
    }
    """

    try:
        data = client.query(query)
        print(f"Found {len(data['dataControllers'])} data controllers")
    except Exception as e:
        print(f"Error: {e}")

    # Async type-safe client
    async with DxtraAsyncClient(os.getenv("DXTRA_API_KEY")) as async_client:
        # Query with variables
        mutation = """
        mutation InsertDataController($object: dataControllers_insert_input!) {
          insertDataController(object: $object) {
            id
            title
            did
          }
        }
        """

        variables = {
            "object": {
                "title": "Example Corp",
                "email": "privacy@example.com"
            }
        }

        result = await async_client.query(mutation, variables)
        controller = result["insertDataController"]
        print(f"Created controller: {controller['id']}")

if __name__ == "__main__":
    asyncio.run(main())

Error Handling Patterns¶

HTTP Status Codes¶

Handling HTTP Errors

import requests

def safe_graphql_request(client, query, variables=None):
    """GraphQL request with comprehensive error handling."""
    try:
        response = client.query(query, variables)
        return {"success": True, "data": response}

    except requests.exceptions.HTTPError as e:
        status_code = e.response.status_code

        if status_code == 401:
            return {"success": False, "error": "Unauthorized - token expired or invalid"}
        elif status_code == 403:
            return {"success": False, "error": "Forbidden - insufficient permissions"}
        elif status_code == 429:
            return {"success": False, "error": "Rate limit exceeded - retry after delay"}
        else:
            return {"success": False, "error": f"HTTP {status_code}: {str(e)}"}

    except Exception as e:
        return {"success": False, "error": f"Request failed: {str(e)}"}

GraphQL Error Responses¶

GraphQL Error Handling

def execute_with_error_handling(client, query, variables=None):
    """Execute GraphQL with detailed error reporting."""
    try:
        # Make request
        headers = {
            "Authorization": f"Bearer {client.jwt_token}",
            "X-Hasura-Role": "user"
        }
        response = requests.post(
            client.api_url,
            headers=headers,
            json={"query": query, "variables": variables or {}}
        )

        # Parse response
        data = response.json()

        # Check for GraphQL errors
        if "errors" in data:
            error = data["errors"][0]
            print(f"GraphQL Error: {error['message']}")

            # Extract error details
            if "extensions" in error:
                print(f"Error code: {error['extensions'].get('code')}")
                print(f"Error path: {error.get('path')}")

            return None

        return data["data"]

    except Exception as e:
        print(f"Request exception: {e}")
        return None

Authentication Guide - Detailed authentication setup
GraphQL Reference - Complete API schema
Error Handling - Comprehensive error patterns