from typing import List from fastapi import APIRouter, FastAPI class RouteRegisterController: def __init__(self, app: FastAPI, router_list: List[APIRouter]): self.router_list = router_list self.app = app @staticmethod def add_router_with_event_to_database(router: APIRouter): from Schemas import EndpointRestriction with EndpointRestriction.new_session() as db_session: EndpointRestriction.set_session(db_session) for route in router.routes: route_path = str(getattr(route, "path")) route_summary = str(getattr(route, "name")) operation_id = getattr(route, "operation_id", None) if not operation_id: raise ValueError(f"Route {route_path} operation_id is not found") if not getattr(route, "methods") and isinstance(getattr(route, "methods")): raise ValueError(f"Route {route_path} methods is not found") route_method = [method.lower() for method in getattr(route, "methods")][0] add_or_update_dict = dict( endpoint_method=route_method, endpoint_name=route_path, endpoint_desc=route_summary.replace("_", " "), endpoint_function=route_summary, is_confirmed=True ) if to_save_endpoint := EndpointRestriction.query.filter(EndpointRestriction.operation_uu_id == operation_id).first(): to_save_endpoint.update(**add_or_update_dict) to_save_endpoint.save() else: created_endpoint = EndpointRestriction.create(**add_or_update_dict, operation_uu_id=operation_id) created_endpoint.save() def register_routes(self): for router in self.router_list: self.app.include_router(router) self.add_router_with_event_to_database(router) return self.app