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.
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 titledescription: API descriptionversion: 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)