Build Autonomous Factory Inspection Agents with CrewAI and PydanticAI

                      
                     
"""
Production implementation for building autonomous factory inspection agents.
Integrates CrewAI for data processing and Pydantic for validation.
"""
from typing import List, Dict, Any, Optional
import os
import logging
import requests
from pydantic import BaseModel, ValidationError
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import time

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

# Database configuration
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///factory_inspections.db')
engine = create_engine(DATABASE_URL)
Base = declarative_base()
SessionLocal = sessionmaker(bind=engine)

class Config:
    """Configuration class for environment variables."""
    database_url: str = DATABASE_URL

class InspectionData(BaseModel):
    """Model for inspection data input."""
    id: int
    status: str
    inspector_id: int
    location: str

class InspectionRecord(Base):
    """Database model for inspection records."""
    __tablename__ = 'inspections'
    id = Column(Integer, primary_key=True, index=True)
    status = Column(String)
    inspector_id = Column(Integer)
    location = Column(String)

Base.metadata.create_all(bind=engine)  # Create tables

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
    """
    try:
        InspectionData(**data)  # Validate using Pydantic
        return True
    except ValidationError as e:
        logger.error(f'Validation error: {e}')
        raise ValueError('Invalid input data')

def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize the input fields to prevent SQL injection.
    
    Args:
        data: Input data
    Returns:
        Sanitized data
    """
    for key in data:
        data[key] = str(data[key]).strip()  # Sanitize each field
    return data

def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """Fetch data from external API.
    
    Args:
        api_url: The API endpoint
    Returns:
        List of dictionaries containing data
    Raises:
        ConnectionError: If API request fails
    """
    try:
        response = requests.get(api_url)
        response.raise_for_status()  # Raise HTTPError for bad responses
        return response.json()
    except requests.RequestException as e:
        logger.error(f'API request failed: {e}')
        raise ConnectionError('Failed to fetch data from API')

def save_to_db(inspection: InspectionData, db: Session) -> None:
    """Save inspection data to the database.
    
    Args:
        inspection: Inspection data to save
        db: Database session
    """
    db_record = InspectionRecord(
        id=inspection.id,
        status=inspection.status,
        inspector_id=inspection.inspector_id,
        location=inspection.location
    )
    db.add(db_record)  # Add record to the session
    db.commit()  # Commit the changes
    db.refresh(db_record)  # Refresh to get the latest data

def process_batch(data_batch: List[Dict[str, Any]]) -> None:
    """Process a batch of inspection data.
    
    Args:
        data_batch: Batch of inspection data
    """
    db = SessionLocal()  # Create a new database session
    try:
        for data in data_batch:
            sanitized_data = sanitize_fields(data)  # Sanitize fields
            if validate_input(sanitized_data):  # Validate data
                inspection = InspectionData(**sanitized_data)
                save_to_db(inspection, db)  # Save to DB
    except Exception as e:
        logger.error(f'Error processing batch: {e}')
    finally:
        db.close()  # Ensure the DB session is closed

def aggregate_metrics(status: str) -> int:
    """Aggregate metrics based on inspection status.
    
    Args:
        status: Status to aggregate
    Returns:
        Count of inspections with the given status
    """
    db = SessionLocal()  # Create a new database session
    try:
        count = db.query(InspectionRecord).filter(InspectionRecord.status == status).count()
        logger.info(f'Aggregated {count} inspections with status: {status}')
        return count
    finally:
        db.close()  # Ensure the DB session is closed

def handle_errors(func):
    """Decorator for handling errors in functions.
    
    Args:
        func: Function to decorate
    Returns:
        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__}: {e}')
            return None
    return wrapper

@handle_errors
def main(api_url: str) -> None:
    """Main function to run the inspection agent.
    
    Args:
        api_url: The API endpoint to fetch inspection data
    """
    data_batch = fetch_data(api_url)  # Fetch data from API
    process_batch(data_batch)  # Process the data batch
    # Example of aggregating metrics
    status_count = aggregate_metrics('completed')  # Count completed inspections
    logger.info(f'Total completed inspections: {status_count}')  # Log the result

if __name__ == '__main__':
    # Example usage
    main('https://api.example.com/inspections')