fastmlapi 0.1.2

A FastAPI-based ML worker node library for easy model deployment

FastMLAPI

A FastAPI-based ML model serving library for easy deployment. Create production-ready ML APIs with minimal boilerplate.

Features

• 🚀 Simple API: Extend MLController and implement a few methods
• 🔄 Automatic /predict endpoint: Generated automatically with proper request/response handling
• 🎯 Flexible prediction: Use load_model() for standard models or @prediction decorator for custom logic
• 🔧 Data pipelines: @preprocessing and @postprocessing decorators for clean data flow
• �️ Custom endpoints: Add additional routes with @route decorator
• �📊 Health checks: Built-in /health endpoint
• 📝 Auto documentation: Swagger/OpenAPI docs out of the box
• 🎨 Customizable: Custom request/response Pydantic models supported

Installation

pip install fastmlapi

Or install from source:

pip install -e .

Quick Start

Option 1: Using load_model() (Recommended for sklearn, XGBoost, etc.)

Best for models that have a standard .predict() method:

from fastmlapi import MLController, preprocessing, postprocessing
import joblib
import numpy as np

class MyClassifier(MLController):
    model_name = "my-classifier"
    model_version = "1.0.0"
    
    def load_model(self):
        """Load and return your model. Called once at startup."""
        return joblib.load("model.pkl")
    
    @preprocessing
    def preprocess(self, data: dict) -> np.ndarray:
        """Transform input data for the model."""
        features = data.get("features", [])
        return np.array(features).reshape(1, -1)
    
    @postprocessing  
    def postprocess(self, prediction: np.ndarray) -> dict:
        """Format model output for API response."""
        return {
            "class": int(prediction[0]),
            "label": "positive" if prediction[0] == 1 else "negative"
        }

if __name__ == "__main__":
    MyClassifier().run()

Option 2: Using @prediction Decorator (Recommended for custom inference)

Best for PyTorch, TensorFlow, or any custom prediction logic:

from fastmlapi import MLController, prediction, postprocessing
import torch

class PyTorchModel(MLController):
    model_name = "pytorch-classifier"
    model_version = "1.0.0"
    
    def load_model(self):
        """Load PyTorch model."""
        model = torch.load("model.pt")
        model.eval()
        return model
    
    @prediction
    def run_inference(self, data: dict) -> torch.Tensor:
        """Custom prediction logic - replaces the default predict behavior."""
        with torch.no_grad():
            tensor = torch.tensor(data["features"], dtype=torch.float32)
            return self.model(tensor)
    
    @postprocessing
    def postprocess(self, output: torch.Tensor) -> dict:
        """Convert tensor to JSON-serializable format."""
        probabilities = torch.softmax(output, dim=-1)
        return {
            "class": int(torch.argmax(probabilities)),
            "confidence": float(probabilities.max())
        }

if __name__ == "__main__":
    PyTorchModel().run()

Option 3: No Model Needed (External APIs, rule-based systems)

When you don't need to load a model at all:

from fastmlapi import MLController, prediction
import requests

class ExternalAPIController(MLController):
    model_name = "external-predictor"
    
    # No load_model() needed!
    
    @prediction
    def call_external_service(self, data: dict) -> dict:
        """Call an external ML service."""
        response = requests.post(
            "https://api.example.com/predict",
            json=data,
            timeout=30
        )
        return response.json()

if __name__ == "__main__":
    ExternalAPIController().run()

How It Works

Prediction Pipeline

Request → preprocess() → predict_raw() → postprocess() → Response
                              ↑
                    Uses @prediction method
                    OR model.predict()

1. preprocess(data): Transform raw input into model-ready format
2. predict_raw(preprocessed_data): Run the actual prediction
   • If @prediction decorator is used → calls your decorated method
   • Otherwise → calls self.model.predict()
3. postprocess(prediction): Format output for the API response

Decorators

Decorator	Purpose	Required?
@preprocessing	Mark a method as the preprocessing step	No (defaults to pass-through)
@postprocessing	Mark a method as the postprocessing step	No (defaults to {"result": prediction})
@prediction	Mark a method as the custom prediction function	No (uses model.predict() by default)
@route	Add a custom API endpoint	No

Running the Server

Direct execution

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

With Uvicorn (for development with auto-reload)

# main.py
classifier = MyClassifier()
app = classifier.app

uvicorn main:app --reload --host 0.0.0.0 --port 8000

API Endpoints

Once running, your API provides:

Endpoint	Method	Description
/predict	POST	Run predictions
/health	GET	Health check status
/	GET	API info
/docs	GET	Swagger UI documentation
/redoc	GET	ReDoc documentation

Example Request

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"data": {"features": [1.0, 2.0, 3.0]}}'

Example Response

{
  "success": true,
  "prediction": {
    "class": 1,
    "label": "positive"
  },
  "metadata": {
    "model_name": "my-classifier",
    "model_version": "1.0.0"
  }
}

Advanced Usage

Custom Endpoints with @route Decorator

Add additional API endpoints beyond /predict using the @route decorator:

from fastmlapi import MLController, preprocessing, postprocessing, route
from fastapi import Request
from typing import List

class MyController(MLController):
    model_name = "my-model"
    
    def load_model(self):
        return joblib.load("model.pkl")
    
    @preprocessing
    def preprocess(self, data: dict):
        return np.array(data["features"]).reshape(1, -1)
    
    @postprocessing
    def postprocess(self, prediction):
        return {"class": int(prediction[0])}
    
    # Custom endpoint for batch predictions
    @route("/batch", methods=["POST"], tags=["Batch"], summary="Batch prediction")
    async def batch_predict(self, request: Request):
        """Process multiple predictions in one request."""
        data = await request.json()
        results = []
        for item in data["items"]:
            result = await self.predict(item)
            results.append(result)
        return {"results": results, "count": len(results)}
    
    # Custom endpoint for model info
    @route("/model-info", methods=["GET"], tags=["Info"])
    async def model_info(self):
        """Get detailed model information."""
        return {
            "name": self.model_name,
            "version": self.model_version,
            "features": ["feature1", "feature2", "feature3"],
            "classes": ["negative", "positive"]
        }
    
    # Custom endpoint with path parameters
    @route("/predict/{category}", methods=["POST"], tags=["Prediction"])
    async def predict_by_category(self, request: Request, category: str):
        """Run prediction for a specific category."""
        data = await request.json()
        data["category"] = category
        return await self.predict(data)

if __name__ == "__main__":
    MyController().run()

@route Decorator Parameters

Parameter	Type	Default	Description
path	str	(required)	URL path (e.g., /batch, /analyze)
methods	list	["GET"]	HTTP methods (["GET"], ["POST"], etc.)
response_model	BaseModel	None	Pydantic model for response validation
tags	list	["Custom"]	OpenAPI tags for documentation
summary	str	None	Short description for docs
description	str	None	Detailed description for docs

Alternative: Using add_route() Method

You can also add routes programmatically in __init__:

class MyController(MLController):
    def __init__(self):
        super().__init__()
        self.add_route("/status", self.get_status, methods=["GET"])
        self.add_route("/batch", self.batch_predict, methods=["POST"])
    
    async def get_status(self):
        return {"status": "ready", "model_loaded": self.is_loaded}
    
    async def batch_predict(self, request: Request):
        data = await request.json()
        return {"results": [await self.predict(item) for item in data["items"]]}

Custom Request/Response Models

Define your own Pydantic models for type-safe requests and responses:

from pydantic import BaseModel, Field
from typing import List

class ImageRequest(BaseModel):
    image_url: str = Field(..., description="URL of the image to analyze")
    threshold: float = Field(0.5, ge=0, le=1, description="Detection confidence threshold")

class DetectedObject(BaseModel):
    label: str
    confidence: float
    bbox: List[float]

class ImageResponse(BaseModel):
    objects: List[DetectedObject]
    count: int

class ObjectDetector(MLController):
    model_name = "object-detector"
    request_model = ImageRequest    # Custom request schema
    response_model = ImageResponse  # Custom response schema
    
    def load_model(self):
        return load_yolo_model()
    
    @preprocessing
    def preprocess(self, data: dict):
        # data contains: {"image_url": "...", "threshold": 0.5}
        image = download_image(data["image_url"])
        self.threshold = data["threshold"]
        return image
    
    @prediction
    def detect(self, image):
        detections = self.model(image)
        return [d for d in detections if d.confidence >= self.threshold]
    
    @postprocessing
    def postprocess(self, detections) -> dict:
        return {
            "objects": [
                {"label": d.label, "confidence": d.conf, "bbox": d.bbox}
                for d in detections
            ],
            "count": len(detections)
        }

Accessing the Model in @prediction Methods

When using @prediction, you can still access the loaded model via self.model:

class HybridController(MLController):
    def load_model(self):
        return {"encoder": load_encoder(), "classifier": load_classifier()}
    
    @prediction
    def predict(self, data: dict):
        # Access multiple models
        encoded = self.model["encoder"].transform(data["text"])
        return self.model["classifier"].predict(encoded)

TensorFlow/Keras Example

from fastmlapi import MLController, preprocessing, postprocessing
import tensorflow as tf
import numpy as np

class KerasClassifier(MLController):
    model_name = "keras-classifier"
    
    def load_model(self):
        return tf.keras.models.load_model("model.h5")
    
    @preprocessing
    def preprocess(self, data: dict) -> np.ndarray:
        return np.array(data["features"]).reshape(1, -1)
    
    @postprocessing
    def postprocess(self, prediction: np.ndarray) -> dict:
        class_idx = int(np.argmax(prediction[0]))
        return {
            "class": class_idx,
            "probabilities": prediction[0].tolist()
        }

Configuration Reference

MLController Class Attributes

Attribute	Type	Default	Description
model_name	str	"ml-model"	Name of your model
model_version	str	"1.0.0"	Model version string
title	str	"FastMLAPI"	API title (shown in docs)
description	str	"ML Model Serving API"	API description
api_version	str	"1.0.0"	API version
request_model	BaseModel	PredictionRequest	Custom Pydantic request model
response_model	BaseModel	PredictionResponse	Custom Pydantic response model
enable_health	bool	True	Enable /health endpoint
enable_docs	bool	True	Enable Swagger/OpenAPI docs

run() Method Parameters

Parameter	Type	Default	Description
host	str	"0.0.0.0"	Host to bind to
port	int	8000	Port to bind to
reload	bool	False	Enable auto-reload (dev only)
**uvicorn_kwargs	dict	{}	Additional Uvicorn options

Development

# Clone the repository
git clone https://github.com/yourusername/fastmlapi.git
cd fastmlapi

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run example
python examples/simple_classifier.py

License

MIT License