Add path matches support into an UnwrappedRequest

src/turbosloth/__main__.py

@@ -7,14 +7,14 @@ import uvicorn
 
 from turbosloth import SlothApp
 from turbosloth.interfaces.serialized import SerializedResponse, SerializedRequest
-from turbosloth.internal_types import QTYPE, BTYPE
+from turbosloth.internal_types import QTYPE, BTYPE, PTYPE
 from turbosloth.req_schema import UnwrappedRequest
 
 app = SlothApp()
 
 
 @app.get("/")
-async def index(req: UnwrappedRequest[QTYPE, BTYPE]) -> SerializedResponse:
+async def index(req: UnwrappedRequest[QTYPE, BTYPE, PTYPE]) -> SerializedResponse:
     return SerializedResponse(200, {}, 'Hello, ASGI Router!')
 
 
@@ -24,7 +24,7 @@ class UserIdSchema:
 
 
 @app.get("/user/")
-async def get_user(req: UnwrappedRequest[UserIdSchema, BTYPE]) -> SerializedResponse:
+async def get_user(req: UnwrappedRequest[UserIdSchema, BTYPE, PTYPE]) -> SerializedResponse:
     print(req)
     resp: dict[str, Any] = {'message': f'Hello, User ы {req.query.user_id}!', 'from': 'server', 'echo': req.body}
     return SerializedResponse(200, {}, resp)
@@ -53,15 +53,22 @@ def foo() -> SomeInternalData:
     return s
 
 
-@app.post("/user")
-async def post_user(req: UnwrappedRequest[QTYPE, UserPostSchema], dat: SomeInternalData) -> SerializedResponse:
+@dataclass
+class PTYPESchema:
+    user_id: int
+
+
+@app.post("/user/u{user_id}r")
+async def post_user(req: UnwrappedRequest[QTYPE, UserPostSchema, PTYPESchema],
+                    dat: SomeInternalData) -> SerializedResponse:
     print(req)
     print(dat)
     resp: dict[str, Any] = {
         'message': f'Hello, User {req.body.user_id}!',
         'from': 'server',
         'data': req.body.data,
-        'inj': dat.a
+        'inj': dat.a,
+        'user_id': req.path_matches.user_id
     }
     return SerializedResponse(200, {}, resp)
 

src/turbosloth/app.py

@@ -14,7 +14,7 @@ from .interfaces.serialized.text import TextSerializedResponse
 from .req_schema import UnwrappedRequest
 from .router import Router
 from .types import HandlerType, InternalHandlerType, ContentType
-from .internal_types import Scope, Receive, Send, MethodType, QTYPE, BTYPE
+from .internal_types import Scope, Receive, Send, MethodType, QTYPE, BTYPE, PTYPE
 from breakshaft.convertor import ConvRepo
 
 from .util import parse_content_type
@@ -188,12 +188,12 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
         self.inj_repo = ConvRepo()
 
         @self.inj_repo.mark_injector()
-        def extract_query(req: BasicRequest) -> QTYPE:
+        def extract_query(req: BasicRequest | SerializedRequest) -> QTYPE:
             return req.query
 
         @self.inj_repo.mark_injector()
-        def extract_query(req: SerializedRequest) -> QTYPE:
-            return req.query
+        def extract_path_matches(req: BasicRequest | SerializedRequest) -> PTYPE:
+            return req.path_matches
 
         @self.inj_repo.mark_injector()
         def extract_body(req: SerializedRequest) -> BTYPE:
@@ -227,15 +227,26 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
             if req_schema is None:
                 raise ValueError(f'Unable to find request schema in handler {fn}')
 
-            query_type, body_type = get_args(req_schema)
+            query_type, body_type, path_type = get_args(req_schema)
             q_inflator = None
             b_inflator = None
+            p_inflator = None
 
             fork_with = set()
 
             def none_generator(*args) -> None:
                 return None
 
+            if path_type not in [PTYPE, None, Any]:
+                p_inflator = self.infl_generator.schema_to_inflator(
+                    path_type,
+                    strict_mode_override=False,
+                    from_type_override=QTYPE
+                )
+                fork_with.add(ConversionPoint(p_inflator, path_type, (PTYPE,)))
+            else:
+                fork_with.add(ConversionPoint(none_generator, path_type, (PTYPE,)))
+
             if query_type not in [QTYPE, None, Any]:
                 q_inflator = self.infl_generator.schema_to_inflator(
                     query_type,
@@ -255,11 +266,16 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
             else:
                 fork_with.add(ConversionPoint(none_generator, body_type, (BTYPE,)))
 
-            def construct_unwrap(q, b) -> UnwrappedRequest:
-                print(f'unwrapping {query_type} and {body_type} with {q} and {b}')
-                return UnwrappedRequest(q, b)
+            def construct_unwrap(q, b, p) -> UnwrappedRequest:
+                return UnwrappedRequest(q, b, p)
 
-            fork_with |= {ConversionPoint(construct_unwrap, req_schema, (query_type or QTYPE, body_type or BTYPE,))}
+            fork_with |= {
+                ConversionPoint(
+                    construct_unwrap,
+                    req_schema,
+                    ((query_type or QTYPE), (body_type or BTYPE), (path_type or PTYPE))
+                )
+            }
 
             tmp_repo = self.inj_repo.fork(fork_with)
 

src/turbosloth/interfaces/serialized/base.py

@@ -30,6 +30,10 @@ class SerializedRequest:
     def headers(self) -> CaseInsensitiveDict[str, str]:
         return self.basic.headers
 
+    @property
+    def path_matches(self) -> dict[str, str]:
+        return self.basic.path_matches
+
     @classmethod
     def deserialize(cls, basic: BasicRequest, charset: str) -> SerializedRequest:
         raise NotImplementedError()

src/turbosloth/internal_types.py

@@ -19,3 +19,4 @@ type MethodType = (
 
 type QTYPE = Annotated[dict[str, Any], 'query_params']
 type BTYPE = Annotated[dict[str, Any] | list[Any] | str | None, 'body']
+type PTYPE = Annotated[dict[str, str], 'path_matches']

src/turbosloth/req_schema.py

@@ -1,13 +1,13 @@
 from dataclasses import dataclass
-from typing import Any, TypeVar
-
-from mypy.visitor import Generic
+from typing import Any, TypeVar, Generic
 
 Q = TypeVar('Q')
 B = TypeVar('B')
+P = TypeVar('P')
 
 
 @dataclass
-class UnwrappedRequest(Generic[Q, B]):
+class UnwrappedRequest(Generic[Q, B, P]):
     query: Q
     body: B
+    path_matches: P