Access Token and API Authentication Issues

API requests are failing with authentication errors, access tokens are expiring, or token generation is not working correctly. This typically occurs due to incorrect token configuration, expired tokens, or missing token permissions.

Common Error Messages:

• 401 Unauthorized: Invalid token
• 403 Forbidden: Token lacks required privileges
• Token has expired
• Invalid or malformed JWT token

You Might Be Asking:

• How do I create an API access token?
• Why is my token not working?
• How do I refresh expired tokens?

1. Create personal access token via UI:

1. Login to DataHub
2. Click your profile → Settings
3. Navigate to "Access Tokens"
4. Click "Generate Token"
5. Enter description and expiration
6. Copy the token (shown only once)
7. Store securely

2. Create token via GraphQL:

mutation createAccessToken {
  createAccessToken(
    input: {
      type: PERSONAL
      actorUrn: "urn:li:corpuser:datahub"
      duration: TWO_MONTHS
      name: "API Integration Token"
      description: "Token for automated ingestion"
    }
  ) {
    accessToken
    expiresAt
  }
}

3. Use token in API requests:

# REST API
curl -X POST http://localhost:8080/api/graphql \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"query": "{ me { username } }"}'

# Python SDK
from datahub.emitter.rest_emitter import DatahubRestEmitter

emitter = DatahubRestEmitter(
    gms_server="http://localhost:8080",
    token="YOUR_ACCESS_TOKEN"
)

4. Configure token with appropriate privileges:

# Check token privileges
query {
  me {
    username
    privileges {
      privilege
    }
  }
}

# Grant privileges to token's user if needed
mutation grantPrivileges {
  assignRole(
    input: {
      actorUrn: "urn:li:corpuser:datahub"
      roleUrn: "urn:li:dataHubRole:Editor"
    }
  )
}

5. Handle token expiration:

from datahub.emitter.rest_emitter import DatahubRestEmitter
import requests

class TokenManager:
    def __init__(self, gms_url, username, password):
        self.gms_url = gms_url
        self.username = username
        self.password = password
        self.token = None
        self.expires_at = None

    def get_token(self):
        if self.token and not self.is_expired():
            return self.token

        # Generate new token
        response = requests.post(
            f"{self.gms_url}/api/graphql",
            json={
                "query": """
                    mutation {
                        createAccessToken(input: {
                            type: PERSONAL,
                            duration: ONE_MONTH
                        }) {
                            accessToken
                            expiresAt
                        }
                    }
                """
            },
            auth=(self.username, self.password)
        )

        data = response.json()
        self.token = data['data']['createAccessToken']['accessToken']
        self.expires_at = data['data']['createAccessToken']['expiresAt']

        return self.token

    def is_expired(self):
        import time
        return time.time() * 1000 > self.expires_at

# Usage
token_manager = TokenManager("http://localhost:8080", "datahub", "password")
emitter = DatahubRestEmitter(
    gms_server="http://localhost:8080",
    token=token_manager.get_token()
)

Access tokens inherit the permissions of the user who created them. For service accounts, create a dedicated user with appropriate roles. Rotate tokens regularly and never commit them to version control. Use environment variables or secret management systems.

• Access Tokens
• Authentication Guide