Normalize and Classify Supplier Quality Certifications with DocTR and Haystack

"""
Production implementation for Normalizing and Classifying Supplier Quality Certifications.
Provides secure, scalable operations using DocTR for document processing and Haystack for NLP tasks.
"""
from typing import Dict, Any, List
import os
import logging
import time
from contextlib import contextmanager
from sqlalchemy import create_engine, Column, Integer, String, Sequence
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.exc import SQLAlchemyError

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

# Database configuration
Base = declarative_base()

class Config:
    SQLALCHEMY_DATABASE_URL: str = os.getenv('DATABASE_URL')

# Define database model for certifications
class Certification(Base):
    __tablename__ = 'certifications'
    id = Column(Integer, Sequence('certification_id_seq'), primary_key=True)
    name = Column(String(50))
    category = Column(String(50))

# Create a database session
@contextmanager
def get_db_session() -> Session:
    """Provide a database session for transactions.
    
    Yields:
        Session object
    """
    engine = create_engine(Config.SQLALCHEMY_DATABASE_URL)
    SessionLocal = sessionmaker(bind=engine)
    session = SessionLocal()
    try:
        yield session
    except SQLAlchemyError as e:
        logger.error(f"Database error: {e}")
        session.rollback()
        raise
    finally:
        session.close()

# Validate input data
async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for certifications.
    
    Args:
        data: Input data to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if not isinstance(data, dict):
        raise ValueError('Input data must be a dictionary')
    if 'name' not in data or 'category' not in data:
        raise ValueError('Missing required fields: name or category')
    return True

# Sanitize input fields
def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields to prevent SQL injection.
    
    Args:
        data: Input fields
    Returns:
        Sanitized fields
    """
    return {k: v.strip() for k, v in data.items()}

# Fetch data from an external API (mock implementation)
async def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """Fetch data from external API.
    
    Args:
        api_url: URL of the API
    Returns:
        List of records
    """
    # Mock response
    return [{'name': 'ISO 9001', 'category': 'Quality Management'},
            {'name': 'ISO 14001', 'category': 'Environmental Management'}]

# Normalize data from raw input
def normalize_data(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Normalize certification data.
    
    Args:
        data: Raw certification data
    Returns:
        Normalized data
    """
    normalized = []
    for record in data:
        normalized.append({
            'name': record['name'].title(),
            'category': record['category'].title()
        })
    return normalized

# Save data to the database
async def save_to_db(data: List[Dict[str, Any]]) -> None:
    """Save normalized data to the database.
    
    Args:
        data: Normalized data
    """
    with get_db_session() as session:
        for item in data:
            cert = Certification(name=item['name'], category=item['category'])
            session.add(cert)
        session.commit()
        logger.info(f"Saved {len(data)} records to the database.")

# Main processing function
async def process_batch(api_url: str) -> None:
    """Main function to process batch of certifications.
    
    Args:
        api_url: URL of the API to fetch data
    """
    try:
        raw_data = await fetch_data(api_url)
        if not raw_data:
            logger.warning('No data fetched from API.')
            return
        validated_data = [sanitize_fields(record) for record in raw_data]
        await validate_input(validated_data)
        normalized_data = normalize_data(validated_data)
        await save_to_db(normalized_data)
    except Exception as e:
        logger.error(f"Error processing batch: {e}")

if __name__ == '__main__':
    # Example usage
    api_url = 'https://api.example.com/certifications'
    import asyncio
    asyncio.run(process_batch(api_url))