fastapi-reloader API Reference

def send_reload_signal():
    """Broadcast a reload signal to all connected clients and break their long-polling connections."""
    for subscribers in requests.values():
        for queue in subscribers:
            queue.put_nowait(1)

@reload_router.head("")
async def heartbeat():
    return Response(status_code=202)

@reload_router.get("")
async def subscribe():
    async def event_generator():
        key = get_id()
        queue = Queue[Literal[0, 1]]()

        stopped = False

        async def heartbeat():
            while not stopped:
                queue.put_nowait(0)
                await sleep(1)

        requests[key].append(queue)

        heartbeat_future = ensure_future(heartbeat())

        try:
            yield "0\n"
            while True:
                value = await queue.get()
                yield f"{value}\n"
                if value == 1:
                    break
        finally:
            heartbeat_future.cancel()
            requests[key].remove(queue)

    return StreamingResponse(event_generator(), 201, media_type="text/plain")

def is_streaming_response(response: Response) -> TypeGuard[StreamingResponse]:
    # In fact, it may not be a Starlette's StreamingResponse, but an internal one with the same interface
    return hasattr(response, "body_iterator")

async def _injection_http_middleware(request: Request, call_next: Callable[[Request], Awaitable[Response]]):
    res = await call_next(request)

    if request.scope.get(FLAG) or request.method != "GET" or "html" not in (res.headers.get("content-type", "")) or res.headers.get("content-encoding", "identity") != "identity":
        return res

    request.scope[FLAG] = True

    async def response():
        if is_streaming_response(res):
            async for chunk in res.body_iterator:
                yield chunk
        else:
            yield res.body

        yield INJECTION

    headers = {k: v for k, v in res.headers.items() if k.lower() not in {"content-length", "transfer-encoding"}}

    return StreamingResponse(response(), res.status_code, headers, res.media_type)

def _wrap_asgi_app(app: ASGIApp):
    @asynccontextmanager
    async def lifespan(_):
        async with LifespanManager(app, inf, inf):
            yield

    new_app = FastAPI(openapi_url=None, lifespan=lifespan)
    new_app.include_router(reload_router)
    new_app.mount("/", app)

    return new_app

def patch_for_auto_reloading(app: ASGIApp):  # this function is preserved for backward compatibility
    if isinstance(app, Starlette):  # both FastAPI and Starlette have user_middleware attribute
        new_app = copy(app)
        new_app.user_middleware = [*app.user_middleware, html_injection_middleware]  # before compression middlewares

        # OTEL patches the app's build_middleware_stack method and keep a reference to the original build_middleware_stack.
        # But both methods are bound to the original app instance, so we need to rebind them to the new app instance.
        # The following loop generically rebinds all these methods, preventing potential issues caused by similar patches.
        for i in dir(new_app):
            if ismethod(method := getattr(new_app, i)) and method.__self__ is app:
                setattr(new_app, i, MethodType(method.__func__, new_app))

        return _wrap_asgi_app(new_app)

    new_app = _wrap_asgi_app(app)
    new_app.user_middleware.append(html_injection_middleware)  # the last middleware is the first one to be called

    return new_app