Add BasePath and Permission

lilya/_internal/iterables.py

@@ -0,0 +1,20 @@
+from typing import Any, Iterator
+
+
+class BaseWrapper:
+    """
+    Builds a wrapper middleware for all the classes.
+    """
+
+    def __init__(self, cls_obj: type, **kwargs: Any) -> None:
+        self.cls_obj = cls_obj
+        self.kwargs = kwargs
+
+    def __iter__(self) -> Iterator[Any]:
+        return iter((self.cls_obj, self.kwargs))
+
+    def __repr__(self) -> str:
+        class_name = self.__class__.__name__
+        option_strings = [f"{key}={value!r}" for key, value in self.kwargs.items()]
+        args_repr = ", ".join([self.cls_obj.__name__] + option_strings)
+        return f"{class_name}({args_repr})"

lilya/app.py

@@ -8,6 +8,7 @@
 from lilya.conf.exceptions import FieldException
 from lilya.conf.global_settings import Settings
 from lilya.datastructures import State
+from lilya.permissions.base import Permission
 from lilya.types import ApplicationType, ASGIApp, ExceptionHandler, Lifespan
 
 
@@ -40,7 +41,7 @@ def __init__(
         routes: Union[Sequence[Any], None] = None,
         middleware: Union[Sequence[Any], None] = None,
         exception_handlers: Union[Mapping[Any, ExceptionHandler], None] = None,
-        permissions: Union[Sequence[Any], None] = None,
+        permissions: Union[Sequence[Permission], None] = None,
         on_startup: Sequence[Callable[[], Any]] | None = None,
         on_shutdown: Sequence[Callable[[], Any]] | None = None,
         lifespan: Optional[Lifespan[ApplicationType]] = None,

lilya/compat.py

@@ -22,6 +22,9 @@ def md5_hexdigest(data: bytes, *, usedforsecurity: bool = True) -> str:
 
 
 def is_async_callable(obj: Any) -> Any:
+    """
+    Validates if a given object is an async callable or not.
+    """
     while isinstance(obj, functools.partial):
         obj = obj.func
 
@@ -74,13 +77,3 @@ def handle_lifespan_events(
     elif lifespan:
         return lifespan
     return None
-
-
-def generate_lifespan_events(
-    on_startup: Union[Sequence[Any], None] = None,
-    on_shutdown: Union[Sequence[Any], None] = None,
-    lifespan: Union[Lifespan[Any], None] = None,
-) -> Any:  # pragma: no cover
-    if lifespan:
-        return lifespan
-    return AyncLifespanContextManager(on_startup=on_startup, on_shutdown=on_shutdown)

lilya/enums.py

@@ -58,3 +58,20 @@ class Event(BaseEnum):
     HTTP_DISCONNECT = "http.disconnect"
     WEBSOCKET_CONNECT = "websocket.connect"
     WEBSOCKET_DISCONNECT = "websocket.disconnect"
+
+
+class Match(IntEnum):
+    NONE = 0
+    PARTIAL = 1
+    FULL = 2
+
+
+class HTTPMethod(BaseEnum):
+    GET = "GET"
+    POST = "POST"
+    PUT = "PUT"
+    PATCH = "PATCH"
+    DELETE = "DELETE"
+    HEAD = "HEAD"
+    OPTIONS = "OPTIONS"
+    TRACE = "TRACE"

lilya/middleware/base.py

@@ -1,25 +1,15 @@
-from typing import Any, Callable, Iterator
+from typing import Any, Callable
 
+from lilya._internal.iterables import BaseWrapper
 from lilya.types import ASGIApp
 
 
-class Middleware:
+class Middleware(BaseWrapper):
     """
     Builds a wrapper middleware for all the classes.
     """
 
-    def __init__(self, cls_obj: type, **kwargs: Any) -> None:
-        self.cls_obj = cls_obj
-        self.kwargs = kwargs
-
-    def __iter__(self) -> Iterator[Any]:
-        return iter((self.cls_obj, self.kwargs))
-
-    def __repr__(self) -> str:
-        class_name = self.__class__.__name__
-        option_strings = [f"{key}={value!r}" for key, value in self.kwargs.items()]
-        args_repr = ", ".join([self.cls_obj.__name__] + option_strings)
-        return f"{class_name}({args_repr})"
+    ...
 
 
 class CreateMiddleware:

lilya/permissions/base.py

@@ -0,0 +1,9 @@
+from lilya._internal.iterables import BaseWrapper
+
+
+class Permission(BaseWrapper):
+    """
+    Builds a wrapper permission for all the classes.
+    """
+
+    ...

lilya/routing.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import inspect
+from typing import Any, Callable, List, Sequence, Set, Tuple, Union
+
+from lilya.datastructures import URLPath
+from lilya.enums import HTTPMethod, Match
+from lilya.middleware.base import Middleware
+from lilya.permissions.base import Permission
+from lilya.types import Receive, Scope, Send
+
+
+def get_name(handler: Callable[..., Any]) -> str:
+    """
+    Returns the name of a given handler.
+    """
+    return (
+        handler.__name__
+        if inspect.isroutine(handler) or inspect.isclass(handler)
+        else handler.__class__.__name__
+    )
+
+
+class BasePath:
+    """
+    The base of all paths (routes) for any ASGI application
+    with Lilya.
+    """
+
+    def search(self, scope: Scope) -> Tuple[Match, Scope]:
+        """
+        Searches for a matching route.
+        """
+        raise NotImplementedError()  # pragma: no cover
+
+    def path_for(self, name: str, /, **path_params: Any) -> URLPath:
+        """
+        Returns a URL of a matching route.
+        """
+        raise NotImplementedError()  # pragma: no cover
+
+    async def dispatch(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """
+        Handles the matched ASGI route.
+        """
+        raise NotImplementedError()  # pragma: no cover
+
+
+class Path(BasePath):
+    """
+    The way you can define a route in Lilya and apply the corresponding
+    path definition.
+
+    ## Example
+
+    ```python
+    from lilya.routing import Path
+
+    Path('/home', callable=..., name="home")
+    ```
+    """
+
+    def __init__(
+        self,
+        path: str,
+        handler: Callable[..., Any],
+        *,
+        methods: Union[List[str], None] = None,
+        name: Union[str, None] = None,
+        include_in_schema: bool = True,
+        middleware: Union[Sequence[Middleware], None] = None,
+        permissions: Union[Sequence[Permission], None] = None,
+    ) -> None:
+        assert path.startswith("/"), "Paths must start with '/'"
+        self.path = path
+        self.handler = handler
+        self.name = get_name(handler) if name is None else name
+        self.include_in_schema = include_in_schema
+        self.methods: Union[List[str], Set[str], None] = methods
+
+        # Defition of the app
+        self.app = handler
+
+        # Execute the middlewares
+        if middleware is not None:
+            for cls, options in reversed(middleware):
+                self.app = cls(app=self.app, **options)
+
+        # Execute the permissions
+        if permissions is not None:
+            for cls, options in reversed(permissions):
+                self.app = cls(app=self.app, **options)
+
+        if self.methods is not None:
+            self.methods = {method.upper() for method in methods}
+            if HTTPMethod.GET in self.methods:
+                self.methods.add(HTTPMethod.HEAD)