Add fastapi endpoint

Author: thomasmeissnercrm

README.md

@@ -197,6 +197,7 @@ using synthetic data that demonstrate BlueCast's full feature set:
 | [07_fairness.py](examples/07_fairness.py) | Fairness auditing, demographic parity, equalized odds, conformal fairness |
 | [08_eda.py](examples/08_eda.py) | Univariate/bivariate plots, PCA, t-SNE, correlations, data quality, leakage detection |
 | [09_bluecast_ai.py](examples/09_bluecast_ai.py) | Multi-agent LLM-powered AutoML (requires API key) |
+| [10_serving.py](examples/10_serving.py) | Deploy models as REST APIs, export Dockerfile |
 
 ### Kaggle competition results
 

bluecast/serve/__init__.py

@@ -0,0 +1,74 @@
+"""
+BlueCast Serve: One-command API deployment for trained pipelines.
+
+Optional module -- install dependencies with:
+    pip install bluecast[serve]
+
+Usage::
+
+    from bluecast.serve import serve, export_api
+
+    # Quick local server
+    serve(automl, port=8080)
+
+    # Export standalone deployment
+    export_api(automl, output_dir="./deployment")
+"""
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def serve(
+    pipeline: Any,
+    host: str = "0.0.0.0",
+    port: int = 8080,
+    log_level: str = "info",
+) -> None:
+    """Start a FastAPI server to serve predictions from a trained pipeline.
+
+    :param pipeline: A trained BlueCast pipeline (any variant).
+    :param host: Host address to bind to.
+    :param port: Port to listen on.
+    :param log_level: Uvicorn log level.
+    """
+    try:
+        import uvicorn
+    except ImportError:
+        raise ImportError(
+            "uvicorn is required to run the server. "
+            "Install with: pip install 'bluecast[serve]'"
+        )
+
+    from bluecast.serve.app import create_app
+
+    app = create_app(pipeline)
+
+    print(f"Starting BlueCast Model API on http://{host}:{port}")
+    print(f"Swagger docs: http://{host}:{port}/docs")
+    print(f"Health check: http://{host}:{port}/health")
+
+    uvicorn.run(app, host=host, port=port, log_level=log_level)
+
+
+def export_api(
+    pipeline: Any,
+    output_dir: str,
+    include_docker: bool = True,
+) -> str:
+    """Export a trained pipeline as a standalone FastAPI deployment.
+
+    Creates a directory with ``app.py``, ``model.pkl``, ``requirements.txt``,
+    an optional ``Dockerfile``, and ``README.md``. The generated app is
+    self-contained and does not depend on ``bluecast.serve`` at runtime.
+
+    :param pipeline: A trained BlueCast pipeline (any variant).
+    :param output_dir: Path to the output directory.
+    :param include_docker: Whether to generate a Dockerfile.
+    :returns: Absolute path to the output directory.
+    """
+    from bluecast.serve.exporter import export_api as _export
+
+    return _export(pipeline, output_dir, include_docker=include_docker)

bluecast/serve/app.py

@@ -0,0 +1,146 @@
+"""FastAPI app factory for serving BlueCast pipelines."""
+
+import logging
+import traceback
+from typing import Any, Dict, List
+
+import numpy as np
+import pandas as pd
+
+from bluecast.serve.schemas import (
+    _get_class_problem,
+    _has_conformal,
+    build_request_model,
+    build_schema_response,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _format_prediction(raw_result: Any, class_problem: str) -> Dict[str, Any]:
+    """Format a single prediction result into a JSON-friendly dict."""
+    if class_problem in ("binary", "multiclass"):
+        if isinstance(raw_result, tuple) and len(raw_result) == 2:
+            probs, classes = raw_result
+            prob_val = probs.tolist() if hasattr(probs, "tolist") else probs
+            class_val = classes.tolist() if hasattr(classes, "tolist") else classes
+            if isinstance(prob_val, list) and len(prob_val) == 1:
+                prob_val = prob_val[0]
+            if isinstance(class_val, list) and len(class_val) == 1:
+                class_val = class_val[0]
+            return {"probabilities": prob_val, "predicted_class": class_val}
+        else:
+            val = raw_result.tolist() if hasattr(raw_result, "tolist") else raw_result
+            return {"prediction": val}
+    else:
+        if isinstance(raw_result, (np.ndarray, pd.Series)):
+            val = raw_result.tolist()
+            if isinstance(val, list) and len(val) == 1:
+                val = val[0]
+            return {"prediction": val}
+        return {"prediction": raw_result}
+
+
+def _format_batch_prediction(raw_result: Any, class_problem: str) -> Dict[str, Any]:
+    """Format batch predictions."""
+    if class_problem in ("binary", "multiclass"):
+        if isinstance(raw_result, tuple) and len(raw_result) == 2:
+            probs, classes = raw_result
+            return {
+                "probabilities": probs.tolist() if hasattr(probs, "tolist") else list(probs),
+                "predicted_classes": (
+                    classes.tolist() if hasattr(classes, "tolist") else list(classes)
+                ),
+                "count": len(probs) if hasattr(probs, "__len__") else 1,
+            }
+    if isinstance(raw_result, (np.ndarray, pd.Series)):
+        return {"predictions": raw_result.tolist(), "count": len(raw_result)}
+    return {"predictions": raw_result}
+
+
+def create_app(pipeline: Any) -> Any:
+    """Create a FastAPI application from a trained BlueCast pipeline.
+
+    The app auto-generates request schemas from the pipeline's column metadata
+    and provides prediction, health, schema, and metrics endpoints.
+
+    :param pipeline: A trained BlueCast pipeline (any variant).
+    :returns: A FastAPI application instance.
+    """
+    try:
+        from fastapi import FastAPI, HTTPException
+    except ImportError:
+        raise ImportError(
+            "FastAPI is required for bluecast.serve. "
+            "Install with: pip install 'bluecast[serve]'"
+        )
+
+    class_problem = _get_class_problem(pipeline)
+    has_conformal = _has_conformal(pipeline)
+
+    RequestModel = build_request_model(pipeline)
+
+    app = FastAPI(
+        title="BlueCast Model API",
+        description=(
+            f"Auto-generated API for a BlueCast {class_problem} model. "
+            f"Conformal prediction: {'enabled' if has_conformal else 'disabled'}."
+        ),
+        version="1.0.0",
+    )
+
+    @app.get("/health")
+    def health() -> Dict[str, Any]:
+        return {
+            "status": "healthy",
+            "model_type": class_problem,
+            "conformal_prediction": has_conformal,
+        }
+
+    @app.get("/schema")
+    def schema() -> Dict[str, Any]:
+        return build_schema_response(pipeline)
+
+    @app.get("/metrics")
+    def metrics() -> Dict[str, Any]:
+        eval_metrics = getattr(pipeline, "eval_metrics", None)
+        if hasattr(pipeline, "_inner"):
+            eval_metrics = getattr(pipeline._inner, "eval_metrics", eval_metrics)
+        if eval_metrics is None:
+            return {"message": "No evaluation metrics available. Use fit_eval() to generate."}
+        serializable = {}
+        for k, v in eval_metrics.items():
+            if isinstance(v, (int, float, str, bool)):
+                serializable[k] = v
+            elif isinstance(v, (np.floating, np.integer)):
+                serializable[k] = float(v)
+        return serializable
+
+    @app.post("/predict")
+    def predict(request: RequestModel) -> Dict[str, Any]:  # type: ignore[valid-type]
+        try:
+            data = request.model_dump()
+            data = {k: v for k, v in data.items() if k != "_placeholder"}
+            df = pd.DataFrame([data])
+            result = pipeline.predict(df)
+            return _format_prediction(result, class_problem)
+        except Exception as e:
+            logger.error(f"Prediction failed: {e}\n{traceback.format_exc()}")
+            raise HTTPException(status_code=400, detail=str(e))
+
+    @app.post("/predict/batch")
+    def predict_batch(requests: List[RequestModel]) -> Dict[str, Any]:  # type: ignore[valid-type]
+        try:
+            data_list = []
+            for req in requests:
+                d = req.model_dump()
+                d = {k: v for k, v in d.items() if k != "_placeholder"}
+                data_list.append(d)
+            df = pd.DataFrame(data_list)
+            result = pipeline.predict(df)
+            return _format_batch_prediction(result, class_problem)
+        except Exception as e:
+            logger.error(f"Batch prediction failed: {e}\n{traceback.format_exc()}")
+            raise HTTPException(status_code=400, detail=str(e))
+
+    return app

bluecast/serve/exporter.py

@@ -0,0 +1,94 @@
+"""Export a trained BlueCast pipeline as a standalone deployment directory."""
+
+import json
+import logging
+import os
+from typing import Any
+
+from bluecast.general_utils.general_utils import save_to_production
+from bluecast.serve.schemas import _extract_column_info, _get_class_problem
+
+logger = logging.getLogger(__name__)
+
+
+def export_api(
+    pipeline: Any,
+    output_dir: str,
+    include_docker: bool = True,
+) -> str:
+    """Export a trained pipeline as a self-contained FastAPI deployment.
+
+    Creates a directory with a standalone ``app.py``, serialized model,
+    ``requirements.txt``, optional ``Dockerfile``, and a ``README.md``.
+    The generated app does NOT depend on ``bluecast.serve`` at runtime —
+    only on ``bluecast`` core, ``fastapi``, and ``uvicorn``.
+
+    :param pipeline: A trained BlueCast pipeline (any variant).
+    :param output_dir: Path to the output directory (created if needed).
+    :param include_docker: Whether to generate a Dockerfile.
+    :returns: The absolute path to the output directory.
+    """
+    try:
+        from jinja2 import Environment, PackageLoader
+    except ImportError:
+        raise ImportError(
+            "jinja2 is required for export_api. "
+            "It should already be installed as a BlueCast dependency."
+        )
+
+    os.makedirs(output_dir, exist_ok=True)
+    abs_output = os.path.abspath(output_dir)
+
+    columns = _extract_column_info(pipeline)
+    class_problem = _get_class_problem(pipeline)
+
+    # Serialize the pipeline (save_to_production appends file_type to path)
+    model_base = os.path.join(abs_output, "model")
+    save_to_production(pipeline, model_base, file_type=".pkl")
+    logger.info(f"Model saved to {model_base}.pkl")
+
+    env = Environment(
+        loader=PackageLoader("bluecast.serve", "templates"),
+        keep_trailing_newline=True,
+    )
+
+    # Render app.py
+    app_template = env.get_template("app_standalone.py.j2")
+    columns_json = json.dumps(columns, indent=4)
+    app_code = app_template.render(
+        columns=columns,
+        columns_json=columns_json,
+        class_problem=class_problem,
+    )
+    _write(abs_output, "app.py", app_code)
+
+    # Render requirements.txt
+    requirements = "bluecast>=3.0.0\nfastapi>=0.100.0\nuvicorn>=0.20.0\ndill>=0.3.3\n"
+    _write(abs_output, "requirements.txt", requirements)
+
+    # Render Dockerfile
+    if include_docker:
+        dockerfile_template = env.get_template("Dockerfile.j2")
+        _write(abs_output, "Dockerfile", dockerfile_template.render())
+
+    # Render README
+    example_payload = json.dumps(
+        {col["name"]: 0.0 if col["python_type"] == "float" else "example" for col in columns[:5]},
+        indent=2,
+    )
+    readme_template = env.get_template("README.md.j2")
+    readme = readme_template.render(
+        class_problem=class_problem,
+        example_payload=example_payload,
+    )
+    _write(abs_output, "README.md", readme)
+
+    logger.info(f"Deployment exported to {abs_output}")
+    return abs_output
+
+
+def _write(directory: str, filename: str, content: str) -> None:
+    path = os.path.join(directory, filename)
+    with open(path, "w") as f:
+        f.write(content)
+    logger.info(f"  Written: {path}")