Environment Variables

Python tasks in Datablast receive date information and configuration through environment variables, providing a clean and secure way to pass data between the platform and your Python code.

Overview

Environment variables in Python tasks provide:

• Date information: Execution dates and time intervals
• Configuration data: Task-specific settings and parameters
• Secret management: Secure access to credentials and API keys
• Platform metadata: Information about the current execution context

Date Environment Variables

Basic Date Variables

import os
from datetime import datetime

# Access date variables through environment variables
data_interval_start = os.getenv('BLAST_DATA_INTERVAL_START')
data_interval_end = os.getenv('BLAST_DATA_INTERVAL_END')
start_date = os.getenv('BLAST_START_DATE')
end_date = os.getenv('BLAST_END_DATE')
start_date_nodash = os.getenv('BLAST_START_DATE_NODASH')
end_date_nodash = os.getenv('BLAST_END_DATE_NODASH')

print(f"Processing data from {start_date} to {end_date}")

Available Date Variables

Variable	Description	Example
BLAST_DATA_INTERVAL_START	Start of data interval	2024-01-15T00:00:00+00:00
BLAST_DATA_INTERVAL_END	End of data interval	2024-01-16T00:00:00+00:00
BLAST_START_DATE	Data interval start date	2024-01-15
BLAST_END_DATE	Data interval end date	2024-01-16
BLAST_START_DATE_NODASH	Start date without dashes	20240115
BLAST_END_DATE_NODASH	End date without dashes	20240116

Working with Date Variables

import os
from datetime import datetime

# Get execution date
execution_date = os.getenv('BLAST_START_DATE')

# Convert to datetime object if needed
if execution_date:
    start_dt = datetime.strptime(execution_date, '%Y-%m-%d')
    print(f"Processing data for {start_dt.strftime('%B %d, %Y')}")

# Use in data processing
def process_data_for_date(date_str):
    """Process data for a specific date."""
    if not date_str:
        raise ValueError("Date string is required")

    # Your data processing logic here
    print(f"Processing data for {date_str}")
    return f"Processed data for {date_str}"

# Call the function
result = process_data_for_date(execution_date)

Configuration Environment Variables

Secret Management

Using Secrets

import os

# Access secrets through environment variables
api_key = os.getenv('API_KEY')
database_password = os.getenv('DB_PASSWORD')
encryption_key = os.getenv('ENCRYPTION_KEY')

# Use secrets in your code
if api_key:
    headers = {'Authorization': f'Bearer {api_key}'}
    # Make API calls
else:
    raise ValueError("API key is required")

Secret Configuration

name: "external_data.api_fetch"
type: "python"
run: "api_fetch.py"
secrets:
  - "API_KEY:API_KEY"
  - "DB_PASSWORD:DB_PASSWORD"
  - "ENCRYPTION_KEY:ENCRYPTION_KEY"

Secure Secret Handling

import os
import logging

logger = logging.getLogger(__name__)

def get_secret(secret_name, required=True):
    """Safely get a secret from environment variables."""
    secret_value = os.getenv(secret_name)

    if required and not secret_value:
        logger.error(f"Required secret {secret_name} is not available")
        raise ValueError(f"Required secret {secret_name} is not available")

    if secret_value:
        logger.info(f"Secret {secret_name} is available")
    else:
        logger.warning(f"Secret {secret_name} is not available")

    return secret_value

# Use the secure function
api_key = get_secret('API_KEY', required=True)
optional_secret = get_secret('OPTIONAL_SECRET', required=False)

Platform Metadata

Execution Context

import os

# Get platform metadata
platform_version = os.getenv('BLAST_PLATFORM_VERSION')
instance_type = os.getenv('BLAST_INSTANCE_TYPE')
region = os.getenv('BLAST_REGION')

print(f"Platform version: {platform_version}")
print(f"Instance type: {instance_type}")
print(f"Region: {region}")

Best Practices

1. Environment Variable Handling

• Always check if environment variables exist before using them
• Provide default values for optional variables
• Use type conversion for numeric values
• Handle missing required variables gracefully

2. Security

• Never log secret values
• Use secure secret handling functions
• Validate secret values before use
• Implement proper error handling

3. Configuration Management

• Use environment variables for configuration
• Provide sensible defaults
• Document required and optional variables
• Use consistent naming conventions

4. Error Handling

• Check for required environment variables
• Provide meaningful error messages
• Log configuration issues appropriately
• Handle type conversion errors

Common Patterns

Configuration Class

import os
from typing import Optional

class TaskConfig:
    """Configuration class for Python tasks."""

    def __init__(self):
        self.execution_date = os.getenv('BLAST_START_DATE')
        self.api_key = os.getenv('API_KEY')
        self.database_url = os.getenv('DATABASE_URL')

    def validate(self):
        """Validate required configuration."""
        if not self.execution_date:
            raise ValueError("Execution date is required")

        if not self.api_key:
            raise ValueError("API key is required")


# Use the configuration class
config = TaskConfig()
config.validate()

print(f"Processing data for {config.execution_date}")

Date Processing Utility

import os
from datetime import datetime, timedelta

def get_date_range():
    """Get the date range for the current execution."""
    start_date_str = os.getenv('BLAST_START_DATE')
    end_date_str = os.getenv('BLAST_END_DATE')

    if not start_date_str:
        raise ValueError("Start date is required")

    start_date = datetime.strptime(start_date_str, '%Y-%m-%d')

    if end_date_str:
        end_date = datetime.strptime(end_date_str, '%Y-%m-%d')
    else:
        end_date = start_date + timedelta(days=1)

    return start_date, end_date

# Use the utility function
start_date, end_date = get_date_range()
print(f"Processing data from {start_date.date()} to {end_date.date()}")

Troubleshooting

Common Issues

Missing Environment Variables

• Issue: Environment variables not available
• Solution: Check task configuration and secret setup
• Debug: Log available environment variables

Type Conversion Errors

• Issue: Invalid type conversions
• Solution: Use proper type conversion with error handling
• Debug: Check variable values and types

Secret Access Issues

• Issue: Secrets not accessible
• Solution: Verify secret configuration and permissions
• Debug: Check secret availability and format

Debugging Tips

import os
import logging

logger = logging.getLogger(__name__)

def debug_environment():
    """Debug environment variables."""
    logger.info("Available environment variables:")

    # List all BLAST_* variables
    blast_vars = {k: v for k, v in os.environ.items() if k.startswith('BLAST_')}
    for key, value in blast_vars.items():
        logger.info(f"  {key}: {value}")

    # Check for secrets (without logging values)
    secret_vars = [k for k in os.environ.keys() if k in ['API_KEY', 'DB_PASSWORD', 'ENCRYPTION_KEY']]
    logger.info(f"Available secrets: {secret_vars}")

# Call debug function
debug_environment()

Related Documentation

• Python Tasks
• Secret Management
• Task Configuration
• Python Dependencies