API Generation Module

Automatic REST API generation for ML models using FastAPI, including endpoint creation, input validation, and documentation.

Overview

The API Generation module provides automatic REST API generation for ML models using FastAPI. It includes endpoint creation, input validation, authentication, monitoring, and deployment support.

Basic Usage

from ai_ml_framework.api import APIGenerator

api = APIGenerator('model.pkl')
app = api.generate_api()
api.generate_main_script('main.py')

Advanced Features

api.add_authentication(api_key='secure-key')
api.enable_rate_limiting(100)
api.add_monitoring_middleware()
app = api.generate_api(title="ML API", version="1.0.0")

APIGenerator

class APIGenerator(model_path: str, model_id: str = None)

Automatic REST API generation for ML models.

Key Features

  • FastAPI integration
  • Automatic endpoint creation
  • Input validation
  • Authentication support
  • Rate limiting
  • Monitoring middleware

Methods

generate_api(title: str = "ML Model API", description: str = "Auto-generated API", version: str = "1.0.0") -> FastAPI

Generate FastAPI application.

Parameters:
  • title: API title
  • description: API description
  • version: API version
Returns: FastAPI application
add_authentication(api_key: str)

Add API key authentication.

Parameters:
  • api_key: API key for authentication
enable_rate_limiting(requests_per_minute: int = 60)

Enable rate limiting.

Parameters:
  • requests_per_minute: Requests per minute limit
generate_main_script(output_path: str = "main.py")

Generate main.py script for running API.

Parameters:
  • output_path: Output file path

APIDeployment

class APIDeployment(api_generator, workspace_dir: str = "deployment_workspace")

Advanced API deployment tools.

Key Features

  • Docker deployment
  • Kubernetes support
  • Cloud platform deployment
  • Load balancing
  • Monitoring setup

Methods

deploy_docker(config: DeploymentConfig) -> Dict[str, Any]

Deploy API using Docker.

Parameters:
  • config: Deployment configuration
Returns: Deployment status
deploy_kubernetes(config: DeploymentConfig) -> Dict[str, Any]

Deploy API to Kubernetes.

Parameters:
  • config: Deployment configuration
Returns: Deployment status

Examples

Basic API Generation

python
from ai_ml_framework.api import APIGenerator
import joblib

# Save your trained model
joblib.dump(best_model, 'my_model.pkl')

# Generate API
api_generator = APIGenerator('my_model.pkl', 'my_model')
app = api_generator.generate_api(
    title="My ML Model API",
    description="Auto-generated API for ML model",
    version="1.0.0"
)

# Generate deployment files
api_generator.generate_main_script('api_main.py')
api_generator.generate_requirements('api_requirements.txt')
api_generator.generate_dockerfile('Dockerfile')

print("API generated successfully!")
print("Run: python api_main.py")

API with Authentication

python
# Create API with authentication
api_generator = APIGenerator('model.pkl', 'secure_model')

# Add authentication
api_generator.add_authentication(api_key='secure-key-123')

# Add rate limiting
api_generator.enable_rate_limiting(requests_per_minute=100)

# Add monitoring
api_generator.add_monitoring_middleware()

# Generate API with features
app = api_generator.generate_api(
    title="Secure ML API",
    description="API with authentication and monitoring",
    version="1.1.0"
)

# Generate enhanced files
api_generator.generate_main_script('secure_api.py')
api_generator.generate_requirements('secure_requirements.txt')
api_generator.generate_dockerfile('secure_Dockerfile')

Custom Endpoints

python
# Add custom endpoints
def custom_health_check():
    return {"status": "healthy", "model": "loaded"}

def model_info():
    return {
        "model_type": "classification",
        "features": model.n_features_,
        "classes": model.classes_.tolist()
    }

def batch_predict(data: List[Dict]):
    # Batch prediction logic
    import pandas as pd
    df = pd.DataFrame(data)
    predictions = model.predict(df)
    return {"predictions": predictions.tolist()}

# Add custom endpoints
api_generator.add_custom_endpoint("/health", ["GET"], custom_health_check)
api_generator.add_custom_endpoint("/info", ["GET"], model_info)
api_generator.add_custom_endpoint("/batch_predict", ["POST"], batch_predict)

# Generate API with custom endpoints
app = api_generator.generate_api(
    title="Advanced ML API",
    description="API with custom endpoints",
    version="2.0.0"
)

Deployment

python
from ai_ml_framework.api import APIDeployment

# Create deployment manager
deployment = APIDeployment(api_generator)

# Docker deployment
docker_config = {
    'image_name': 'ml-api',
    'tag': 'latest',
    'port': 8000,
    'environment': {
        'API_KEY': 'secure-key-123',
        'LOG_LEVEL': 'INFO'
    }
}

docker_result = deployment.deploy_docker(docker_config)
print(f"Docker deployment: {docker_result}")

# Kubernetes deployment
k8s_config = {
    'namespace': 'ml-apis',
    'replicas': 3,
    'resources': {
        'requests': {'cpu': '100m', 'memory': '128Mi'},
        'limits': {'cpu': '500m', 'memory': '512Mi'}
    }
}

k8s_result = deployment.deploy_kubernetes(k8s_config)
print(f"Kubernetes deployment: {k8s_result}")

# Cloud deployment
cloud_config = {
    'platform': 'aws',
    'region': 'us-west-2',
    'instance_type': 't2.micro'
}

cloud_result = deployment.deploy_cloud(cloud_config)
print(f"Cloud deployment: {cloud_result}")

Local Testing

python
import requests
import json

# Test API locally
base_url = "http://localhost:8000"

# Test health endpoint
response = requests.get(f"{base_url}/health")
print(f"Health check: {response.json()}")

# Test prediction endpoint
test_data = {
    "data": {
        "feature_1": 1.0,
        "feature_2": 2.0,
        "feature_3": 3.0
    }
}

response = requests.post(
    f"{base_url}/predict",
    json=test_data,
    headers={"X-API-Key": "secure-key-123"}
)
print(f"Prediction: {response.json()}")

# Test batch prediction
batch_data = {
    "data": [
        {"feature_1": 1.0, "feature_2": 2.0, "feature_3": 3.0},
        {"feature_1": 1.5, "feature_2": 2.5, "feature_3": 3.5}
    ]
}

response = requests.post(
    f"{base_url}/batch_predict",
    json=batch_data,
    headers={"X-API-Key": "secure-key-123"}
)
print(f"Batch predictions: {response.json()}")

Generated Files

python
# Generated main.py
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import uvicorn

# Load model
model = joblib.load('my_model.pkl')

# Pydantic models
class PredictionRequest(BaseModel):
    data: dict

class PredictionResponse(BaseModel):
    prediction: float
    confidence: float

# FastAPI app
app = FastAPI(
    title="My ML Model API",
    description="Auto-generated API for ML model",
    version="1.0.0"
)

# Endpoints
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    # Convert data to DataFrame
    import pandas as pd
    df = pd.DataFrame([request.data])
    
    # Make prediction
    prediction = model.predict(df)[0]
    
    # Get confidence (if available)
    confidence = 0.95  # Placeholder
    
    return PredictionResponse(
        prediction=prediction,
        confidence=confidence
    )

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)