Build Real-Time Factory Analytics Agents with Prompt Optimisation Using AdalFlow and FastAPI

"""
Production implementation for building real-time factory analytics agents.
This solution provides secure, scalable operations with prompt optimization using AdalFlow.
"""

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, constr
from typing import List, Dict, Any
import os
import logging
import httpx
import asyncio
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 configuration
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///./test.db')
engine = create_engine(DATABASE_URL)
Base = declarative_base()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

class AnalyticsData(BaseModel):
    """Model for incoming analytics data."""
    factory_id: constr(strip_whitespace=True, min_length=1)
    sensor_id: constr(strip_whitespace=True, min_length=1)
    value: float

class AnalyticsRecord(Base):
    """Database model for analytics records."""
    __tablename__ = 'analytics'
    id = Column(Integer, primary_key=True, index=True)
    factory_id = Column(String, index=True)
    sensor_id = Column(String)
    value = Column(Integer)

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

async def validate_input(data: AnalyticsData) -> bool:
    """Validate the input data.

    Args:
        data: Input AnalyticsData to validate.
    Returns:
        True if valid.
    Raises:
        ValueError: If validation fails.
    """
    if data.value < 0:
        raise ValueError('Sensor value must be non-negative.')  # Validate sensor value
    logger.info('Input data validated')
    return True

async def save_to_db(data: AnalyticsData, db) -> None:
    """Save validated data to the database.

    Args:
        data: Validated AnalyticsData to save.
        db: Database session.
    """
    record = AnalyticsRecord(factory_id=data.factory_id, sensor_id=data.sensor_id, value=data.value)
    db.add(record)
    db.commit()  # Commit the record
    logger.info('Data saved to database')

async def fetch_data(url: str) -> Dict[str, Any]:
    """Fetch data from an external API.

    Args:
        url: API endpoint to fetch data from.
    Returns:
        JSON response from the API.
    Raises:
        HTTPException: If fetching fails.
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            response.raise_for_status()
            logger.info('Data fetched from API')
            return response.json()
    except Exception as e:
        logger.error('Error fetching data: %s', str(e))
        raise HTTPException(status_code=500, detail='Error fetching data')

async def process_batch(data_list: List[AnalyticsData]) -> None:
    """Process a batch of data records.

    Args:
        data_list: List of AnalyticsData to process.
    """
    for data in data_list:
        await validate_input(data)  # Validate each record
        async with SessionLocal() as db:
            await save_to_db(data, db)  # Save each validated record

async def aggregate_metrics(factory_id: str, db) -> Dict[str, float]:
    """Aggregate metrics for a given factory.

    Args:
        factory_id: Factory ID to aggregate metrics for.
        db: Database session.
    Returns:
        Dictionary of aggregated metrics.
    """
    metrics = {'average': 0.0, 'count': 0}
    records = db.query(AnalyticsRecord).filter(AnalyticsRecord.factory_id == factory_id).all()
    if records:
        metrics['count'] = len(records)
        metrics['average'] = sum(record.value for record in records) / metrics['count']
    logger.info('Aggregated metrics: %s', metrics)
    return metrics

app = FastAPI()  # Create FastAPI app

@app.post('/analytics/', response_model=AnalyticsData)
async def receive_data(data: AnalyticsData) -> AnalyticsData:
    """Receive analytics data and process it.

    Args:
        data: AnalyticsData received from the request.
    Returns:
        The processed AnalyticsData.
    Raises:
        HTTPException: If processing fails.
    """
    try:
        await validate_input(data)
        async with SessionLocal() as db:
            await save_to_db(data, db)
        logger.info('Data received and processed')
        return data
    except ValueError as ve:
        logger.error('Value error: %s', str(ve))
        raise HTTPException(status_code=400, detail=str(ve))
    except Exception as e:
        logger.error('Processing error: %s', str(e))
        raise HTTPException(status_code=500, detail='Error processing data')

@app.get('/metrics/{factory_id}', response_model=Dict[str, float])
async def get_metrics(factory_id: str) -> Dict[str, float]:
    """Get aggregated metrics for a specific factory.

    Args:
        factory_id: ID of the factory to get metrics for.
    Returns:
        Aggregated metrics for the factory.
    Raises:
        HTTPException: If fetching metrics fails.
    """
    async with SessionLocal() as db:
        metrics = await aggregate_metrics(factory_id, db)
    return metrics

if __name__ == '__main__':
    # Example usage
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)