Refactored _Routes to Server method

Author: michalpokusa

adafruit_httpserver/route.py

@@ -8,7 +8,7 @@
 """
 
 try:
-    from typing import Callable, List, Iterable, Union, Tuple, Dict, TYPE_CHECKING
+    from typing import Callable, Iterable, Union, Tuple, Literal, Dict, TYPE_CHECKING
 
     if TYPE_CHECKING:
         from .response import Response
@@ -36,13 +36,18 @@ def __init__(
         self.parameters_names = [
             name[1:-1] for name in re.compile(r"/[^<>]*/?").split(path) if name != ""
         ]
-        self.path = re.sub(r"<\w+>", r"([^/]+)", path).replace("....", r".+").replace(
-            "...", r"[^/]+"
-        ) + ("/?" if append_slash else "")
+        self.path_pattern = re.compile(
+            r"^"
+            + re.sub(r"<\w+>", r"([^/]+)", path)
+            .replace(r"....", r".+")
+            .replace(r"...", r"[^/]+")
+            + (r"/?" if append_slash else r"")
+            + r"$"
+        )
+        self.path = path
         self.methods = (
             set(methods) if isinstance(methods, (set, list, tuple)) else set([methods])
         )
-
         self.handler = handler
 
     @staticmethod
@@ -56,7 +61,9 @@ def _validate_path(path: str, append_slash: bool) -> None:
         if path.endswith("/") and append_slash:
             raise ValueError("Cannot use append_slash=True when path ends with /")
 
-    def match(self, other: "Route") -> Tuple[bool, Dict[str, str]]:
+    def matches(
+        self, method: str, path: str
+    ) -> Union[Tuple[Literal[False], None], Tuple[Literal[True], Dict[str, str]]]:
         """
         Checks if the route matches the other route.
 
@@ -68,45 +75,41 @@ def match(self, other: "Route") -> Tuple[bool, Dict[str, str]]:
 
         Examples::
 
-            route = Route("/example", GET, True)
+            route = Route("/example", GET, append_slash=True)
 
-            other1a = Route("/example", GET)
-            other1b = Route("/example/", GET)
-            route.matches(other1a) # True, {}
-            route.matches(other1b) # True, {}
+            route.matches(GET, "/example") # True, {}
+            route.matches(GET, "/example/") # True, {}
 
-            other2 = Route("/other-example", GET)
-            route.matches(other2) # False, {}
+            route.matches(GET, "/other-example") # False, None
+            route.matches(POST, "/example/") # False, None
 
             ...
 
             route = Route("/example/<parameter>", GET)
 
-            other1 = Route("/example/123", GET)
-            route.matches(other1) # True, {"parameter": "123"}
+            route.matches(GET, "/example/123") # True, {"parameter": "123"}
 
-            other2 = Route("/other-example", GET)
-            route.matches(other2) # False, {}
+            route.matches(GET, "/other-example") # False, None
 
             ...
 
-            route1 = Route("/example/.../something", GET)
-            other1 = Route("/example/123/something", GET)
-            route1.matches(other1) # True, {}
+            route = Route("/example/.../something", GET)
+            route.matches(GET, "/example/123/something") # True, {}
 
-            route2 = Route("/example/..../something", GET)
-            other2 = Route("/example/123/456/something", GET)
-            route2.matches(other2) # True, {}
+            route = Route("/example/..../something", GET)
+            route.matches(GET, "/example/123/456/something") # True, {}
         """
 
-        if not other.methods.issubset(self.methods):
-            return False, {}
+        if method not in self.methods:
+            return False, None
+
+        path_match = self.path_pattern.match(path)
+        if path_match is None:
+            return False, None
 
-        regex_match = re.match(f"^{self.path}$", other.path)
-        if regex_match is None:
-            return False, {}
+        url_parameters_values = path_match.groups()
 
-        return True, dict(zip(self.parameters_names, regex_match.groups()))
+        return True, dict(zip(self.parameters_names, url_parameters_values))
 
     def __repr__(self) -> str:
         path = repr(self.path)
@@ -168,51 +171,3 @@ def route_decorator(func: Callable) -> Route:
         return Route(path, methods, func, append_slash=append_slash)
 
     return route_decorator
-
-
-class _Routes:
-    """A collection of routes and their corresponding handlers."""
-
-    def __init__(self) -> None:
-        self._routes: List[Route] = []
-
-    def add(self, route: Route):
-        """Adds a route and its handler to the collection."""
-        self._routes.append(route)
-
-    def find_handler(self, route: Route) -> Union[Callable["...", "Response"], None]:
-        """
-        Finds a handler for a given route.
-
-        If route used URL parameters, the handler will be wrapped to pass the parameters to the
-        handler.
-
-        Example::
-
-            @server.route("/example/<my_parameter>", GET)
-            def route_func(request, my_parameter):
-                ...
-                request.path == "/example/123" # True
-                my_parameter == "123" # True
-        """
-        found_route, _route = False, None
-
-        for _route in self._routes:
-            matches, keyword_parameters = _route.match(route)
-
-            if matches:
-                found_route = True
-                break
-
-        if not found_route:
-            return None
-
-        handler = _route.handler
-
-        def wrapped_handler(request):
-            return handler(request, **keyword_parameters)
-
-        return wrapped_handler
-
-    def __repr__(self) -> str:
-        return f"_Routes({repr(self._routes)})"

adafruit_httpserver/server.py

@@ -30,7 +30,7 @@
 from .methods import GET, HEAD
 from .request import Request
 from .response import Response, FileResponse
-from .route import _Routes, Route
+from .route import Route
 from .status import BAD_REQUEST_400, UNAUTHORIZED_401, FORBIDDEN_403, NOT_FOUND_404
 
 
@@ -65,7 +65,7 @@ def __init__(
         self._auths = []
         self._buffer = bytearray(1024)
         self._timeout = 1
-        self._routes = _Routes()
+        self._routes: "List[Route]" = []
         self._socket_source = socket_source
         self._sock = None
         self.headers = Headers()
@@ -132,7 +132,7 @@ def route_func(request):
         """
 
         def route_decorator(func: Callable) -> Callable:
-            self._routes.add(Route(path, methods, func, append_slash=append_slash))
+            self._routes.append(Route(path, methods, func, append_slash=append_slash))
             return func
 
         return route_decorator
@@ -157,8 +157,7 @@ def add_routes(self, routes: List[Route]) -> None:
                 external_route2,
             ]}
         """
-        for route in routes:
-            self._routes.add(route)
+        self._routes.extend(routes)
 
     def _verify_can_start(self, host: str, port: int) -> None:
         """Check if the server can be successfully started. Raises RuntimeError if not."""
@@ -296,6 +295,35 @@ def _receive_request(
 
         return request
 
+    def _find_handler(
+        self, method: str, path: str
+    ) -> Union[Callable[..., "Response"], None]:
+        """
+        Finds a handler for a given route.
+
+        If route used URL parameters, the handler will be wrapped to pass the parameters to the
+        handler.
+
+        Example::
+
+            @server.route("/example/<my_parameter>", GET)
+            def route_func(request, my_parameter):
+                ...
+                request.path == "/example/123" # True
+                my_parameter == "123" # True
+        """
+        for route in self._routes:
+            route_matches, url_parameters = route.matches(method, path)
+
+            if route_matches:
+
+                def wrapped_handler(request):
+                    return route.handler(request, **url_parameters)
+
+                return wrapped_handler
+
+        return None
+
     def _handle_request(
         self, request: Request, handler: Union[Callable, None]
     ) -> Union[Response, None]:
@@ -371,8 +399,8 @@ def poll(self) -> str:
                 conn.close()
                 return CONNECTION_TIMED_OUT
 
-            # Find a handler for the route
-            handler = self._routes.find_handler(Route(request.path, request.method))
+            # Find a route that matches the request's method and path and get its handler
+            handler = self._find_handler(request.method, request.path)
 
             # Handle the request
             response = self._handle_request(request, handler)