app.py

import logging
import os
import time

import gradio as gr
import structlog
import uvicorn
from asgi_correlation_id import CorrelationIdMiddleware
from asgi_correlation_id.context import correlation_id
from dotenv import load_dotenv
from fastapi import FastAPI, Request, Response
from uvicorn.protocols.utils import get_path_with_query_string

from helpers import formatters
from helpers.structlog_setup import setup_logging
from routes import router


load_dotenv()

# LOG_JSON_FORMAT = parse_obj_as(bool, os.getenv("LOG_JSON_FORMAT", False))
LOG_JSON_FORMAT = bool(os.getenv("LOG_JSON_FORMAT", False))
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
setup_logging(json_logs=LOG_JSON_FORMAT, log_level=LOG_LEVEL)
logger = structlog.stdlib.get_logger(__name__)
title = "Example API"
app = FastAPI(title=title, version="1.0.0")


@app.middleware("http")
async def logging_middleware(request: Request, call_next) -> Response:
    structlog.contextvars.clear_contextvars()
    # These context vars will be added to all log entries emitted during the request
    request_id = correlation_id.get()
    # print(f"request_id:{request_id}.")
    structlog.contextvars.bind_contextvars(request_id=request_id)

    start_time = time.perf_counter_ns()
    # If the call_next raises an error, we still want to return our own 500 response,
    # so we can add headers to it (process time, request ID...)
    response = Response(status_code=500)
    try:
        response = await call_next(request)
    except Exception:
        # TODO: Validate that we don't swallow exceptions (unit test?)
        structlog.stdlib.get_logger("api.error").exception("Uncaught exception")
        raise
    finally:
        process_time = time.perf_counter_ns() - start_time
        status_code = response.status_code
        url = get_path_with_query_string(request.scope)
        client_host = request.client.host
        client_port = request.client.port
        http_method = request.method
        http_version = request.scope["http_version"]
        # Recreate the Uvicorn access log format, but add all parameters as structured information
        logger.info(
            f"""{client_host}:{client_port} - "{http_method} {url} HTTP/{http_version}" {status_code}""",
            http={
                "url": str(request.url),
                "status_code": status_code,
                "method": http_method,
                "request_id": request_id,
                "version": http_version,
            },
            network={"client": {"ip": client_host, "port": client_port}},
            duration=process_time,
        )
        response.headers["X-Process-Time"] = str(process_time / 10 ** 9)
        return response


app.include_router(router)
logger.info("routes included, creating gradio app")
CUSTOM_GRADIO_PATH = "/"


def get_gradio_app():
    with gr.Blocks() as gradio_app:
        logger.info("start gradio app building...")
        gr.Markdown(
            """
            # Hello World!

            Start typing below to _see_ the *output*.

            Here a [link](https://huggingface.co/spaces/aletrn/gradio_with_fastapi).
            """
        )
        btn = gr.Button(value="Divide et Impera")
        text_input = gr.Textbox(lines=1, placeholder="10",
                                label="write an integer to divide 100; 0 will raise ZeroDivisionError")
        text_output = gr.Textbox(lines=1, placeholder=None, label="Text Output")
        gr.Examples(
            examples=[19, 1, -7, 0],
            inputs=[text_input],
        )
        btn.click(
            formatters.request_formatter,
            inputs=[text_input, ],
            outputs=[text_output]
        )
    return gradio_app


logger.info("mounting gradio app within FastAPI...")
gradio_app_md = get_gradio_app()
app.add_middleware(CorrelationIdMiddleware)
app = gr.mount_gradio_app(app, gradio_app_md, path=CUSTOM_GRADIO_PATH)
logger.info("gradio app mounted")

if __name__ == "__main__":
    try:
        uvicorn.run(host="0.0.0.0", port=7860, app=app)
    except Exception as ex:
        logger.error(f"fastapi/gradio application {title}, exception:{ex}.")
        print(f"fastapi/gradio application {title}, exception:{ex}.")
        raise ex