Build Stream-to-Iceberg Table Pipelines for Factory Analytics with Redpanda and DuckDB

"""
Production implementation for building stream-to-Iceberg table pipelines.
Provides secure, scalable operations for factory analytics.
"""

from typing import Dict, Any, List, Optional
import os
import logging
import json
import time
import requests
import duckdb
from redpanda import Producer, Consumer

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class to manage environment variables.
    """
    database_url: str = os.getenv('DATABASE_URL')
    redpanda_broker: str = os.getenv('REDPANDA_BROKER')

def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'id' not in data:
        raise ValueError('Missing id')  # ID is mandatory
    if 'value' not in data:
        raise ValueError('Missing value')  # Value is mandatory
    return True

def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields to prevent injection attacks.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized data dictionary
    """
    sanitized = {k: str(v).strip() for k, v in data.items()}
    logger.info('Sanitized fields: %s', sanitized)
    return sanitized

def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform records for Iceberg table compatibility.
    
    Args:
        records: List of records to transform
    Returns:
        Transformed records
    """
    transformed = []
    for record in records:
        transformed.append({
            'id': record['id'],
            'value': float(record['value']),
            'timestamp': int(time.time())
        })  # Convert timestamp to Unix time
    logger.info('Transformed records: %s', transformed)
    return transformed

def fetch_data(topic: str, consumer: Consumer) -> List[Dict[str, Any]]:
    """Fetch data from Redpanda topic.
    
    Args:
        topic: The Redpanda topic to consume from
        consumer: Redpanda consumer instance
    Returns:
        List of consumed records
    """
    logger.info('Fetching data from topic: %s', topic)
    records = []
    for message in consumer:
        records.append(json.loads(message.value.decode('utf-8')))
        if len(records) >= 100:  # Limit number of records
            break
    return records

def save_to_db(records: List[Dict[str, Any]], db_connection: duckdb.DuckDBPyConnection) -> None:
    """Save transformed records to DuckDB.
    
    Args:
        records: List of records to save
        db_connection: DuckDB connection instance
    """
    try:
        for record in records:
            db_connection.execute(
                "INSERT INTO iceberg_table (id, value, timestamp) VALUES (?, ?, ?)" ,
                (record['id'], record['value'], record['timestamp'])
            )
        logger.info('Records saved to DuckDB successfully.')
    except Exception as e:
        logger.error('Failed to save records: %s', str(e))
        raise

def handle_errors(func):
    """Decorator to handle errors in processing.
    """    
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error('Error occurred: %s', str(e))
            raise
    return wrapper

class DataPipeline:
    """
    Main orchestrator for the data pipeline.
    """
    def __init__(self, config: Config):
        self.config = config
        self.consumer = Consumer(self.config.redpanda_broker)
        self.db_connection = duckdb.connect(self.config.database_url)

    @handle_errors
    def run(self) -> None:
        """Run the data pipeline.
        """
        topic = 'factory-data'
        logger.info('Starting data pipeline...')
        records = fetch_data(topic, self.consumer)
        logger.info('Fetched %d records.', len(records))
        for record in records:
            validate_input(record)
            sanitized = sanitize_fields(record)
            transformed = transform_records([sanitized])
            save_to_db(transformed, self.db_connection)
        logger.info('Data pipeline completed successfully.')

    def __del__(self):
        """Cleanup resources on deletion.
        """
        self.db_connection.close()
        logger.info('Database connection closed.')

if __name__ == '__main__':
    config = Config()
    pipeline = DataPipeline(config)
    pipeline.run()  # Start the pipeline