Fine-Tune SmolLM3 for Structured Equipment Diagnostics with Unsloth and Instructor

"""
Production implementation for fine-tuning SmolLM3 for structured equipment diagnostics.
Provides secure, scalable operations with improved data handling.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import httpx
import asyncio
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# Logger setup for tracking application behavior
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class for environment variables and application settings
class Config:
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///./test.db')  # Fallback to SQLite for testing

# SQLAlchemy setup for database interactions
Base = declarative_base()
engine = create_engine(Config.database_url, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class EquipmentData(Base):
    """
    SQLAlchemy model for equipment data storage.
    
    Attributes:
        id: Unique identifier for equipment
        name: Equipment name
        status: Current status of equipment
    """
    __tablename__ = "equipment"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    status = Column(String)

# Ensure the database tables are created
Base.metadata.create_all(bind=engine)

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for equipment.
    
    Args:
        data: Input data dictionary.
    Returns:
        bool: True if valid, raises ValueError otherwise.
    Raises:
        ValueError: If any required field is missing.
    """
    if 'name' not in data:
        raise ValueError('Missing required field: name')  # Validate presence of 'name'
    return True

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields for security.
    
    Args:
        data: Input data dictionary.
    Returns:
        Dict[str, Any]: Sanitized data dictionary.
    """
    return {key: str(value).strip() for key, value in data.items()}  # Strip whitespace

async def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize input data for processing.
    
    Args:
        data: Input data dictionary.
    Returns:
        Dict[str, Any]: Normalized data dictionary.
    """
    return {**data, 'status': data['status'].lower()}  # Normalize status to lowercase

async def transform_records(data: List[Dict[str, Any]]) -> List[EquipmentData]:
    """Transform raw input data into EquipmentData instances.
    
    Args:
        data: List of raw input data dictionaries.
    Returns:
        List[EquipmentData]: List of EquipmentData instances.
    """
    return [EquipmentData(name=record['name'], status=record['status']) for record in data]

async def fetch_data(url: str) -> List[Dict[str, Any]]:
    """Fetch data from an external API.
    
    Args:
        url: API endpoint to fetch data from.
    Returns:
        List[Dict[str, Any]]: List of records fetched from API.
    Raises:
        Exception: If request fails.
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()  # Raise for HTTP errors
        return response.json()  # Return JSON data

async def save_to_db(records: List[EquipmentData], db: Session) -> None:
    """Save list of EquipmentData to the database.
    
    Args:
        records: List of EquipmentData instances to save.
        db: SQLAlchemy session object.
    """
    db.add_all(records)  # Add all records to the session
    db.commit()  # Commit the transaction

async def aggregate_metrics(records: List[EquipmentData]) -> Dict[str, Any]:
    """Aggregate metrics from equipment data.
    
    Args:
        records: List of EquipmentData instances.
    Returns:
        Dict[str, Any]: Aggregated metrics.
    """
    total_count = len(records)
    statuses = [record.status for record in records]
    return {'total_count': total_count, 'statuses': statuses}  # Return aggregated metrics

async def handle_errors(func):
    """Handle errors in asynchronous functions. Decorator for retry logic.
    """  
    async def wrapper(*args, **kwargs):
        for attempt in range(3):  # Retry logic with 3 attempts
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                logger.error(f'Error occurred: {e}')  # Log the error
                await asyncio.sleep(2 ** attempt)  # Exponential backoff
    return wrapper

class SmolLM3FineTuner:
    """Main orchestrator class for fine-tuning SmolLM3.
    
    Attributes:
        db: SQLAlchemy session.
    """
    def __init__(self) -> None:
        self.db: Session = SessionLocal()  # Initialize the database session

    async def process(self, api_url: str) -> None:
        """Main processing workflow for fine-tuning.
        
        Args:
            api_url: API URL to fetch data from.
        """
        try:
            raw_data = await fetch_data(api_url)  # Fetch data from API
            validated = await validate_input(raw_data)  # Validate data
            sanitized = await sanitize_fields(raw_data)  # Sanitize input
            normalized = await normalize_data(sanitized)  # Normalize fields
            records = await transform_records(normalized)  # Transform to EquipmentData
            await save_to_db(records, self.db)  # Save to database
            metrics = await aggregate_metrics(records)  # Aggregate metrics
            logger.info(f'Processing completed: {metrics}')  # Log completion info
        except Exception as e:
            logger.error(f'Processing failed: {e}')  # Log any processing errors
        finally:
            self.db.close()  # Ensure database session is closed

if __name__ == '__main__':
    # Example usage of the SmolLM3FineTuner class
    api_url = 'https://api.example.com/equipment'  # Replace with the actual API URL
    tuner = SmolLM3FineTuner()  # Create an instance of the fine-tuner
    asyncio.run(tuner.process(api_url))  # Run the main processing workflow