#!/usr/bin/env python3 # -*- coding: utf-8 -*- import re from typing import Self, Any from Interfaces.Application.AnPInterface import AnPInterface from Models.RouteModel import RouteModel from Models.RequestModel import RequestModel from Utils.Common import Common from Utils.Checks import Check class RoutesManager: def __init__(self:Self, anp:AnPInterface) -> None: self.anp:AnPInterface = anp self.__routes:list[RouteModel] = [] self.update() def update(self:Self) -> None: key:str for key in ("default_routes_files", "routes_files", "default_routes", "routes"): self.add(self.anp.settings.get(key), True) def reset(self:Self) -> None: self.__routes = [] self.update() def go(self:Self, method:str, path:str, request:RequestModel) -> None: route:RouteModel for route in self.__routes: if route.match(method, path, request): if not request.response: if route.callback: route.callback(request) elif route.path: index:str for index in [""] + self.anp.indexes.get(): path_indexed:str|None = Common.get_absolute_path(route.path + "/" + path + ("/" + index if index else "")) if path_indexed is not None and Common.is_file(path_indexed): request.set_response(Common.load_file(path_indexed, "rb")) request.response_mime = Common.get_mime_from_path(path_indexed) return request.set_response({ "ok" : False, "code" : 404, "message" : "not_found" }) elif route.controller and route.controller_method: if not self.anp.controllers.execute(route.controller, route.controller_method, request): request.set_response({ "ok" : False, "code" : 505, "message" : "not_implemented" }) else: request.set_response({ "ok" : True, "code" : 500, "message" : "internal_server_error" }) return request.set_response({ "ok" : False, "code" : 404, "message" : "not_found" }) def __add_new_route(self:Self, new_route:RouteModel, overwrite:bool = False) -> bool: if new_route.error: return False i:int route:RouteModel for i, route in enumerate(self.__routes): if route.path == new_route.path and route.method == new_route.method: if overwrite: self.__routes[i] = new_route return True self.__routes.append(new_route) return True def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None: subinputs:Any|None for subinputs in Common.load_json(inputs, False): if isinstance(subinputs, RouteModel): self.__add_new_route(subinputs, overwrite) elif ( Check.is_string(subinputs) or Check.is_array(subinputs) or Check.is_dictionary(subinputs) ) and not self.__add_new_route(RouteModel(subinputs), overwrite): if Check.is_string(subinputs): self.add(subinputs, overwrite) elif Check.is_array(subinputs): fragment:Any|None for fragment in subinputs: if Check.is_string(fragment) and not self.__add_new_route(RouteModel(fragment), overwrite): continue self.add(fragment, overwrite)