Middleware

(request, call_next) → response | call_next(request)

flowchart LR
  Req[Request] --> Global[Global middleware]
  Global --> Route[Route middleware]
  Route --> Handler[API handler]
  Handler --> Res[Response]

"""Entry point: register routes, middleware, and start the server."""

import src.modules.products.products  # registers @route classes

from fusion_framework.app import FusionApp
from fusion_framework.config import get_settings, load_settings_module

# Global middleware (optional). Framework ships with none by default.
MIDDLEWARE: list = []


def main() -> None:
    load_settings_module("settings")
    app = FusionApp(get_settings())
    for middleware in MIDDLEWARE:
        app.use(middleware)
    app.listen()


if __name__ == "__main__":
    main()

def logging_middleware(request, call_next):
    print(request["method"], request["path"])
    return call_next(request)

MIDDLEWARE = [logging_middleware]

def require_api_key(request, call_next):
    key = (request.get("headers") or {}).get("x-api-key")
    if key != "secret":
        return {"status": 401, "body": {"detail": "invalid api key"}}
    return call_next(request)

def get(self):
    user = self.state.get("user")
    return self.response({"user": user})

from fusion_framework.route import route

def only_post(request, call_next):
    if request.get("method", "").upper() != "POST":
        return {"status": 405, "body": {"detail": "POST only"}}
    return call_next(request)

@route("api/[module]/", middleware=[only_post])
class UploadModule(FusionBaseApi):
    def post(self):
        return self.response({"ok": True})

from fusion_framework import bearer_jwt

MIDDLEWARE = [bearer_jwt()]
# or with your verifier:
# MIDDLEWARE = [bearer_jwt(verify=my_verify_fn)]

@route("api/admin", roles=["admin", "super_admin"])
class AdminModule(FusionBaseApi):
    def get(self):
        return self.response({"ok": True})

from fusion_framework.middleware import require_roles

@route("api/admin", middleware=[require_roles("admin", "super_admin")])

import asyncio

async def slow_check(request, call_next):
    await asyncio.sleep(0.01)
    return await call_next(request)

MIDDLEWARE = [slow_check]