Temporal Worker Authentication

from application_sdk.clients.temporal import TemporalWorkflowClient

# Initialize client with authentication
client = TemporalWorkflowClient(
    application_name="postgres-extraction",
    # Auth settings read from environment variables
)

# Establish authenticated connection
await client.load()

# Create and run worker
worker = client.create_worker(
    activities=[my_activity],
    workflow_classes=[MyWorkflow],
)
await worker.run()

from application_sdk.clients.temporal import TemporalWorkflowClient

client = TemporalWorkflowClient(application_name="my-app")
await client.load()

# Access the auth client
auth_client = client.auth_manager

# Get current token
token = await auth_client.get_access_token()

# Force token refresh
new_token = await auth_client.get_access_token(force_refresh=True)

# Get token expiry information
expiry_time = auth_client.get_token_expiry_time()
time_until_expiry = auth_client.get_time_until_expiry()

# Get authenticated headers for HTTP requests
headers = await auth_client.get_authenticated_headers()

# Use with external API calls
import aiohttp
async with aiohttp.ClientSession() as session:
    response = await session.get(
        "https://api.company.com/data",
        headers=headers
    )

import asyncio
from typing import Optional

async def connect_with_retry(
    client: TemporalWorkflowClient,
    max_retries: int = 3,
    base_delay: float = 2.0
) -> None:
    """Connect to Temporal with retry logic.
    
    Args:
        client: The workflow client
        max_retries: Maximum number of retry attempts
        base_delay: Base delay for exponential backoff (seconds)
    """
    for attempt in range(max_retries):
        try:
            await client.load()
            logger.info("Successfully connected to Temporal")
            return
        except Exception as e:
            if attempt < max_retries - 1:
                delay = base_delay ** attempt
                logger.warning(
                    f"Connection attempt {attempt + 1} failed: {e}. "
                    f"Retrying in {delay}s..."
                )
                
                # Clear auth cache before retry
                if hasattr(client, 'auth_manager'):
                    client.auth_manager.clear_cache()
                
                await asyncio.sleep(delay)
            else:
                logger.error(f"Failed to connect after {max_retries} attempts")
                raise

import logging

# Enable debug logging for auth client
logging.getLogger("application_sdk.clients.atlan_auth").setLevel(logging.DEBUG)

# Enable debug logging for secret store
logging.getLogger("application_sdk.services.secretstore").setLevel(logging.DEBUG)

# Test Dapr secret access
dapr invoke \
  --app-id your-app \
  --method get-secret \
  --data '{"key": "atlan-deployment-secrets"}'

# Check Dapr components
kubectl get components

# View component configuration
kubectl get component deployment-secret-store -o yaml

import jwt
import json

# Decode token (without verification) to inspect claims
token = await auth_client.get_access_token()
decoded = jwt.decode(token, options={"verify_signature": False})
print(json.dumps(decoded, indent=2))