Track Domain Fine-Tuning Experiments Across Factory Datasets with LlamaFactory and Weights and Biases

"""
Production implementation for tracking domain fine-tuning experiments across factory datasets using LlamaFactory and Weights and Biases.
Provides secure, scalable operations with robust error handling and logging.
"""
from typing import Dict, Any, List
import os
import logging
import time
import requests
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

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

# Database base model for SQLAlchemy
Base = declarative_base()

class Experiment(Base):
    """SQLAlchemy model for storing experiment results."""
    __tablename__ = 'experiments'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    status = Column(String)
    metrics = Column(String)  # JSON string for metrics

class Config:
    """Configuration class for environment variables."""
    database_url: str = os.getenv('DATABASE_URL')
    api_key: str = os.getenv('API_KEY')

# Create a SQLAlchemy engine and session factory
engine = create_engine(Config.database_url)
Session = sessionmaker(bind=engine)


def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for experiments.
    
    Args:
        data: Input data to validate.
    Returns:
        bool: True if valid, raises ValueError otherwise.
    Raises:
        ValueError: If validation fails.
    """
    if 'name' not in data:
        raise ValueError('Missing experiment name')
    if 'metrics' not in data:
        raise ValueError('Missing metrics data')
    return True

def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize input data for processing.
    
    Args:
        data: Raw input data.
    Returns:
        Dict: Normalized data.
    """
    data['metrics'] = str(data['metrics'])  # Ensure metrics are string
    return data

def fetch_data(api_url: str) -> Dict[str, Any]:
    """Fetch data from an external API.
    
    Args:
        api_url: URL of the API to fetch data from.
    Returns:
        Dict: Response data from the API.
    Raises:
        Exception: If the API request fails.
    """
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        logger.error(f'Error fetching data from {api_url}: {str(e)}')
        raise Exception('Failed to fetch data')

def save_to_db(data: Dict[str, Any]) -> None:
    """Save experiment data to the database.
    
    Args:
        data: Experiment data to save.
    Raises:
        Exception: If database operation fails.
    """
    session = Session()
    try:
        experiment = Experiment(name=data['name'], status='completed', metrics=data['metrics'])
        session.add(experiment)
        session.commit()
        logger.info(f'Experiment {data["name"]} saved successfully.')
    except Exception as e:
        logger.error(f'Error saving to database: {str(e)}')
        session.rollback()
        raise Exception('Failed to save experiment data')
    finally:
        session.close()  # Ensure session is closed

def aggregate_metrics(metrics: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate metrics from multiple experiments.
    
    Args:
        metrics: List of metrics dictionaries.
    Returns:
        Dict: Aggregated metrics.
    """
    aggregated = {}
    for metric in metrics:
        for key, value in metric.items():
            if key in aggregated:
                aggregated[key] += value
            else:
                aggregated[key] = value
    return aggregated

def handle_errors(func):
    """Decorator to handle errors for functions.
    
    Args:
        func: Function to wrap.
    Returns:
        Callable: Wrapped function with error handling.
    """
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {str(e)}')
            raise
    return wrapper

class ExperimentManager:
    """Manager for handling experiments and their lifecycle."""
    def __init__(self) -> None:
        self.session = Session()  # Initialize session

    @handle_errors
    def run_experiment(self, api_url: str) -> None:
        """Run an experiment based on API data.
        
        Args:
            api_url: URL to fetch the experiment data from.
        """
        raw_data = fetch_data(api_url)  # Fetch raw data
        if validate_input(raw_data):  # Validate fetched data
            normalized_data = normalize_data(raw_data)  # Normalize data
            save_to_db(normalized_data)  # Save to DB
            logger.info('Experiment run successfully.')  # Log success

if __name__ == '__main__':
    # Example usage
    manager = ExperimentManager()  # Create manager instance
    api_url = 'https://api.example.com/experiments'
    manager.run_experiment(api_url)  # Run experiment