Source code for glados.router

import logging
from typing import Callable, Dict, NoReturn

from .errors import GladosRouteNotFoundError
from .request import GladosRequest
from .route_type import RouteType


[docs]class GladosRoute(object): """Represents a single route""" def __init__(self, route_type: RouteType, route: str, function: Callable): self.route_type = route_type self.route = route self.function = function def __repr__(self): return f"<GladosRoute| route: {self.route} | route type: {self.route_type.name} | function: {self.function.__name__} >"
[docs]class GladosRouter(object): """GladosRouter""" def __init__(self, **kwargs): # routes are stored as: {RouteType.SendMessage: {"ask_user",ask_user, "confirm":confirm}} self.routes = dict() # type: Dict[RouteType, Dict[str, Callable]] for route in RouteType._member_names_: self.routes[RouteType[route].value] = dict() # type: Dict[str, Callable]
[docs] def add_route(self, plugin, route: GladosRoute) -> NoReturn: """Add a route to the router Parameters ---------- plugin: :obj: `GladosPlugin` the plugin the route belongs to route the route to be added Raises ------ KeyError a route with the same type and same name already exists """ logging.debug(f"adding route: {route}") if route.route in self.routes[route.route_type.value]: raise KeyError( f"a route with the name of {route.route} already exists in the route type: {route.route_type.name}" ) self.routes[route.route_type.value][route.route] = plugin.send_request
[docs] def add_routes(self, plugin) -> NoReturn: """Add multiple routes to the router. Parameters ---------- plugin: :obj: `GladosPlugin` the plugin to add routes from """ for route in plugin.routes: self.add_route(plugin, route)
[docs] def get_route(self, route_type: RouteType, route: str) -> Callable: """Get a GladosRoute object for the requested route. Parameters ---------- route_type the type of route to get route the route to get Raises ------ GladosRouteNotFoundError the requested route is not found """ if not self.routes[route_type.value].get(route): raise GladosRouteNotFoundError( f"no route with the name of {route} exists in route type: {route_type.name}" ) return self.routes[route_type.value][route]
[docs] def route_function(self, route_type: RouteType, route: str) -> Callable: """Return only the callable function for the requested GladosRoute. Parameters ---------- route_type the type of route to get route the route to get Returns ------- Callable return the requested routes callable function """ return self.get_route(route_type, route)
[docs] def exec_route(self, request: GladosRequest): """Execute a route function directly Parameters ---------- request the GLaDOS request Returns ------- the data returned by the plugin Examples ---------- >>> def mock_function(request: GladosRequest): ... print(f"Mock Function: {request.params.message}") ... return True >>> router = GladosRouter() >>> route = GladosRoute(RouteType.SendMessage, "send_mock", mock_function) >>> router.add_route(route) >>> request = GladosRequest(RouteType.SendMessage, "send_mock", message="Hello World!") >>> successful = router.exec_route(request) Mock Function: Hello World! >>> print(successful) True >>> def mock_function(request: GladosRequest): ... print(f"Mock Function: {request.params.message}") ... return True >>> router = GladosRouter() >>> route = GladosRoute(RouteType.SendMessage, "send_mock", mock_function) >>> router.add_route(route) >>> request = GladosRequest(RouteType.SendMessage, "send_mock_fail", message="Hello World!") >>> successful = router.exec_route(request) >>> print(successful) False """ logging.debug(f"calling route function for {request.route}") return self.route_function(request.route_type, request.route)(request)