data-collector-debugtool/app/main.py

from copy import deepcopy
from typing import Any

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from uvicorn.config import LOGGING_CONFIG

from app.api.bacnet import router as bacnet_router
from app.api.modbus import router as modbus_router
from app.api.s7 import router as s7_router
from app.response import response_payload


def uvicorn_log_config() -> dict[str, Any]:
    config = deepcopy(LOGGING_CONFIG)
    config["formatters"]["default"]["fmt"] = "%(asctime)s %(levelprefix)s %(message)s"
    config["formatters"]["default"]["datefmt"] = "%Y-%m-%d %H:%M:%S"
    config["formatters"]["access"]["fmt"] = '%(asctime)s %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s'
    config["formatters"]["access"]["datefmt"] = "%Y-%m-%d %H:%M:%S"
    return config


def validation_error_message(exc: RequestValidationError) -> str:
    messages: list[str] = []
    for error in exc.errors():
        location = ".".join(str(item) for item in error.get("loc", []) if item != "body")
        message = error.get("msg", "validation error")
        messages.append(f"{location}: {message}" if location else message)
    return "; ".join(messages) or "validation error"


def create_app() -> FastAPI:
    app = FastAPI(title="Data Collector Gateway")
    app.include_router(modbus_router, prefix="/api/dc-gateway")
    app.include_router(s7_router, prefix="/api/dc-gateway")
    app.include_router(bacnet_router, prefix="/api/dc-gateway")

    @app.exception_handler(RequestValidationError)
    async def request_validation_exception_handler(
        request: Request,
        exc: RequestValidationError,
    ) -> JSONResponse:
        data: dict[str, list] = {}
        if request.url.path.endswith("/modbus/read"):
            data = {"communication": []}
        elif request.url.path.endswith(("/modbus/read_points", "/modbus/read-points")):
            data = {"points": []}
        elif request.url.path.endswith("/s7/read"):
            data = {"communication": []}
        elif request.url.path.endswith("/s7/read_points"):
            data = {"points": []}
        elif request.url.path.endswith("/s7/connect_scan"):
            data = {"available": []}
        elif request.url.path.endswith(("/bacnet/read_points", "/bacnet/search_points")):
            data = {"points": []}
        elif request.url.path.endswith("/bacnet/bbmd/whois"):
            data = {"devices": []}
        return JSONResponse(
            status_code=200,
            content=response_payload(1, validation_error_message(exc), data),
        )

    return app


app = create_app()


def main() -> None:
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000, log_config=uvicorn_log_config())