Add pattern-substitutive by regexp path matching

src/turbosloth/router.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import re
 import typing
 from typing import Optional, Sequence
 
+from pathspec import Pattern
+
 from .exceptions import MethodNotAllowedException, NotFoundException
 from .types import InternalHandlerType
 from .internal_types import MethodType
@@ -10,31 +13,68 @@ from .internal_types import MethodType
 
 class Route:
     static_subroutes: dict[str, Route]
+    regexp_subroutes: list[tuple[Pattern, Route]]
     handler: dict[MethodType, InternalHandlerType]
 
     def __init__(self) -> None:
         self.static_subroutes = {}
+        self.regexp_subroutes = []
         self.handler = {}
 
+    def _find_regexp_subroute(self, p: Pattern) -> Optional[Route]:
+        for _p, _r in self.regexp_subroutes:
+            if _p == p:
+                return _r
+        return None
+
+    def _add_regexp_subroute(self, p: Pattern, r: Route):
+        if self._find_regexp_subroute(p) is None:
+            self.regexp_subroutes.append((p, r))
+            self.regexp_subroutes.sort(key=lambda x: len(x[0].pattern), reverse=True)
+
     def add(self, method: MethodType, sequence: Sequence[str], handler: InternalHandlerType) -> None:
         if len(sequence) == 0:
             self.handler[method] = handler
             return
 
-        subroute = self.static_subroutes.get(sequence[0])
-        if subroute is None:
-            subroute = Route()
-            self.static_subroutes[sequence[0]] = subroute
+        part = sequence[0]
+
+        if '{' in part:
+            if '}' not in part:
+                raise ValueError(f'Invalid subpath substitute placeholder: {part}')
+            re_part = part.replace('(', '\\(').replace(')', '\\)')
+            re_part = re.sub(r'\{.+}', r'(.+)', re_part)
+            re_part = re.compile('^' + re_part + '$')
+
+            subroute = self._find_regexp_subroute(re_part)
+            if subroute is None:
+                subroute = Route()
+                self._add_regexp_subroute(re_part, subroute)
+        else:
+            subroute = self.static_subroutes.get(part)
+            if subroute is None:
+                subroute = Route()
+                self.static_subroutes[part] = subroute
 
         subroute.add(method, sequence[1:], handler)
 
     def get(self, method: MethodType, sequence: Sequence[str]) -> Optional[InternalHandlerType]:
         if len(sequence) == 0:
+            if len(self.handler) == 0:
+                raise NotFoundException('')
             ret = self.handler.get(method)
             if ret is None:
                 raise MethodNotAllowedException(', '.join(map(str, self.handler.keys())))
             return ret
+
         subroute = self.static_subroutes.get(sequence[0])
+
+        if subroute is None:
+            for p, sr in self.regexp_subroutes:
+                m = p.match(sequence[0])
+                if m is not None:
+                    subroute = sr
+
         if subroute is None:
             raise NotFoundException('/'.join(sequence))
         return subroute.get(method, sequence[1:])

tests/test_router.py

@@ -5,13 +5,13 @@ from src.turbosloth.router import Router
 
 
 def test_router_root_handler():
-    async def f(_: dict, *args):
+    async def f(*args):
         pass
 
-    async def d(_: dict, *args):
+    async def d(*args):
         pass
 
-    async def a(_: dict, *args):
+    async def a(*args):
         pass
 
     r = Router()
@@ -38,10 +38,10 @@ def test_router_root_handler():
 
 
 def test_router_match():
-    async def f(_: dict, *args):
+    async def f(*args):
         pass
 
-    async def d(_: dict, *args):
+    async def d(*args):
         pass
 
     r = Router()
@@ -52,5 +52,30 @@ def test_router_match():
     with pytest.raises(MethodNotAllowedException, match=f'405\tMethod Not Allowed: POST /asdf, allowed: GET'):
         r.match('POST', 'asdf')
 
-    with pytest.raises(MethodNotAllowedException, match=f'405\tMethod Not Allowed: POST /, allowed: none'):
+    with pytest.raises(NotFoundException, match=f'404\tNot Found'):
         r.match('POST', '')
+
+
+def test_router_pattern_match():
+    async def f(*args):
+        pass
+
+    r = Router()
+    r.add('GET', '/{some}/asdf', f)
+    r.add('GET', '/{some}/b{some1}c', f)
+
+    assert r.match('GET', '/1234/asdf')
+    assert r.match('GET', '/ /asdf')
+    assert r.match('GET', '/ /basdfc')
+
+    with pytest.raises(NotFoundException, match='404\tNot Found: asd'):
+        r.match('GET', 'asd')
+
+    with pytest.raises(NotFoundException, match='404\tNot Found: asd'):
+        r.match('GET', 'asd/')
+
+    with pytest.raises(NotFoundException, match='404\tNot Found: asd'):
+        r.match('GET', 'asd/b')
+
+    with pytest.raises(NotFoundException, match='404\tNot Found: asd'):
+        r.match('GET', 'asd/basdf')