app.py

"""
api_light_hf — FastAPI entrypoint.

Routes: POST /{api_name}
Each api_name maps to apis/<api_name>.py which must export a function named <api_name>.
"""

import importlib
import inspect
import json
import logging
import os
import time
import threading
from typing import Any, Dict

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
)
logger = logging.getLogger("app")

app = FastAPI(
    title="api_light_hf",
    description="API endpoints powered by Hugging Face Inference API",
    version="1.0.0",
)

_active = 0
_lock = threading.Lock()
_start = time.time()


# ---------------------------------------------------------------------------
# API Registry
# ---------------------------------------------------------------------------

def _load_apis() -> Dict[str, Any]:
    """Dynamically import all apis/<name>.py and index the exported function."""
    apis: Dict[str, Any] = {}
    api_dir = os.path.join(os.path.dirname(__file__), "apis")

    for fname in os.listdir(api_dir):
        if not fname.endswith(".py") or fname.startswith("__"):
            continue
        module_name = fname[:-3]
        try:
            module = importlib.import_module(f"apis.{module_name}")
            func = getattr(module, module_name, None)
            if func is None:
                logger.warning(f"[{module_name}] no function named '{module_name}' found — skipped")
                continue
            apis[module_name] = func
            logger.info(f"[{module_name}] loaded")
        except Exception as exc:
            logger.error(f"[{module_name}] load error: {exc}")

    logger.info(f"Loaded {len(apis)} APIs")
    return apis


_APIS: Dict[str, Any] = {}


@app.on_event("startup")
def startup_event():
    global _APIS
    _APIS = _load_apis()


# ---------------------------------------------------------------------------
# Health / Info
# ---------------------------------------------------------------------------

@app.get("/")
def root():
    return {
        "status": "ok",
        "uptime_seconds": round(time.time() - _start, 1),
        "active_requests": _active,
        "endpoints": sorted(_APIS.keys()),
    }


@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/endpoints")
def endpoints():
    result = {}
    for name, func in sorted(_APIS.items()):
        sig = inspect.signature(func)
        params = {
            k: str(v.annotation.__name__ if v.annotation != inspect.Parameter.empty else "any")
            for k, v in sig.parameters.items()
            if k != "request"
        }
        result[name] = {
            "doc": (func.__doc__ or "").strip()[:200],
            "params": params,
        }
    return result


# ---------------------------------------------------------------------------
# Dynamic dispatch: POST /{api_name}
# ---------------------------------------------------------------------------

@app.post("/{api_name}")
async def dispatch(api_name: str, request: Request):
    global _active

    func = _APIS.get(api_name)
    if func is None:
        raise HTTPException(status_code=404, detail=f"API '{api_name}' not found")

    # Parse body
    try:
        body = await request.json()
    except Exception:
        body = {}

    if not isinstance(body, dict):
        raise HTTPException(status_code=400, detail="Request body must be a JSON object")

    # Filter to only accepted params (drop unknown keys)
    sig = inspect.signature(func)
    accepted_params = {
        k for k, v in sig.parameters.items()
        if k != "request"
    }
    filtered = {k: v for k, v in body.items() if k in accepted_params}

    with _lock:
        _active += 1
    try:
        # Some legacy functions accept a `request` kwarg (Gradio Request compat)
        if "request" in sig.parameters:
            result = func(**filtered, request=request)
        else:
            result = func(**filtered)
    except Exception as exc:
        logger.error(f"[{api_name}] execution error: {exc}", exc_info=True)
        raise HTTPException(status_code=500, detail=str(exc))
    finally:
        with _lock:
            _active -= 1

    # Normalise result to JSON-serialisable form
    if hasattr(result, "model_dump"):
        return JSONResponse(result.model_dump())
    if isinstance(result, (dict, list, str, int, float, bool, type(None))):
        return JSONResponse(result)
    # Fallback: try converting
    try:
        return JSONResponse(json.loads(json.dumps(result, default=str)))
    except Exception:
        return JSONResponse({"result": str(result)})


# ---------------------------------------------------------------------------
# Entrypoint (local dev)
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "app:app",
        host="0.0.0.0",
        port=int(os.environ.get("PORT", 7860)),
        reload=False,
    )