Add new lilya support

esmerald/permissions/utils.py

@@ -1,7 +1,10 @@
-from typing import TYPE_CHECKING, Callable, Optional
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
+
+from lilya.permissions.base import DefinePermission
 
 from esmerald.exceptions import PermissionDenied
-from esmerald.utils.helpers import is_async_callable
+from esmerald.utils.helpers import is_async_callable, is_class_and_subclass
 
 if TYPE_CHECKING:  # pragma: no cover
     from esmerald.permissions import BasePermission
@@ -40,3 +43,36 @@ def permission_denied(request: "Request", message: Optional[str] = None) -> None
     If request is not permitted, determine what kind of exception to raise.
     """
     raise PermissionDenied(detail=message, status_code=403)
+
+
+@lru_cache
+def is_esmerald_permission(permission: Union["BasePermission", Any]) -> bool:
+    """
+    Checks if the given permission is an instance or subclass of BasePermission.
+
+    Args:
+        permission (Union["BasePermission", Any]): The permission to check.
+    Returns:
+        bool: True if the permission is an instance or subclass of BasePermission, False otherwise.
+    """
+
+    from esmerald.permissions import BasePermission
+
+    return is_class_and_subclass(permission, BasePermission)
+
+
+def wrap_permission(permission: Union["BasePermission", Any]) -> "BasePermission":
+    """
+    Wraps the given permission into a BasePermission instance if it is not already one.
+    Or else it will assume its a Lilya permission and wraps it.
+
+    Args:
+        permission (Union["BasePermission", Any]): The permission to be wrapped.
+    Returns:
+        BasePermission: The wrapped permission instance.
+    """
+
+    if is_esmerald_permission(permission):
+        return permission
+
+    return DefinePermission(permission)

esmerald/routing/gateways.py

@@ -8,6 +8,7 @@
 from lilya.types import Receive, Scope, Send
 from typing_extensions import Annotated, Doc
 
+from esmerald.permissions.utils import is_esmerald_permission, wrap_permission
 from esmerald.routing.apis.base import View
 from esmerald.routing.base import Dispatcher
 from esmerald.typing import Void, VoidType
@@ -298,6 +299,12 @@ def __init__(
         self._middleware: List["Middleware"] = self.handle_middleware(
             handler=handler, base_middleware=self.middleware
         )
+
+        lilya_permissions = [
+            wrap_permission(permission)
+            for permission in permissions or []
+            if not is_esmerald_permission(permission)
+        ]
         super().__init__(
             path=self.path,
             handler=cast(Callable, handler),
@@ -306,6 +313,7 @@ def __init__(
             methods=self.methods,
             middleware=self._middleware,
             exception_handlers=exception_handlers,
+            permissions=lilya_permissions,
         )
         """
         A "bridge" to a handler and router mapping functionality.
@@ -317,7 +325,9 @@ def __init__(
         self.handler = cast("Callable", handler)
         self.dependencies = dependencies or {}
         self.interceptors: Sequence["Interceptor"] = interceptors or []
-        self.permissions: Sequence["Permission"] = permissions or []  # type: ignore
+        self.permissions: Sequence[Permission] = [
+            permission for permission in permissions or [] if is_esmerald_permission(permission)
+        ]  # type: ignore
         self.response_class = None
         self.response_cookies = None
         self.response_headers = None
@@ -523,12 +533,18 @@ def __init__(
         )
         self.is_middleware: bool = False
 
+        lilya_permissions = [
+            wrap_permission(permission)
+            for permission in permissions or []
+            if not is_esmerald_permission(permission)
+        ]
         super().__init__(
             path=self.path,
             handler=cast("Callable", handler),
             name=name,
             middleware=self._middleware,
             exception_handlers=exception_handlers,
+            permissions=lilya_permissions,
         )
         """
         A "bridge" to a handler and router mapping functionality.
@@ -539,7 +555,9 @@ def __init__(
         self.handler = cast("Callable", handler)
         self.dependencies = dependencies or {}
         self.interceptors = interceptors or []
-        self.permissions = permissions or []  # type: ignore
+        self.permissions: Sequence[Permission] = [
+            permission for permission in permissions or [] if is_esmerald_permission(permission)
+        ]  # type: ignore
         self.include_in_schema = False
         self.parent = parent
         (handler.path_regex, handler.path_format, handler.param_convertors, _) = compile_path(

esmerald/routing/router.py

@@ -56,6 +56,7 @@
 from esmerald.openapi.datastructures import OpenAPIResponse
 from esmerald.openapi.utils import is_status_code_allowed
 from esmerald.params import Form
+from esmerald.permissions.utils import is_esmerald_permission, wrap_permission
 from esmerald.requests import Request
 from esmerald.responses import Response
 from esmerald.routing._internal import OpenAPIFieldInfoMixin
@@ -515,13 +516,20 @@ async def another(request: Request) -> str:
             on_startup is None and on_shutdown is None
         ), "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both."
 
+        lilya_permissions = [
+            wrap_permission(permission)
+            for permission in permissions or []
+            if not is_esmerald_permission(permission)
+        ]
+
         super().__init__(
             redirect_slashes=redirect_slashes,
             routes=new_routes,
             default=default,
             lifespan=lifespan,
             on_shutdown=on_shutdown,
             on_startup=on_startup,
+            permissions=lilya_permissions,
         )
         self.path = path
         self.on_startup = [] if on_startup is None else list(on_startup)
@@ -530,7 +538,9 @@ async def another(request: Request) -> str:
         self.dependencies = dependencies or {}
         self.exception_handlers = exception_handlers or {}
         self.interceptors: Sequence[Interceptor] = interceptors or []
-        self.permissions: Sequence[Permission] = permissions or []  # type: ignore
+        self.permissions: Sequence[Permission] = [
+            permission for permission in permissions or [] if is_esmerald_permission(permission)
+        ]  # type: ignore
         self.routes: Any = new_routes
         self.middleware = middleware or []
         self.tags = tags or []
@@ -1988,12 +1998,20 @@ def __init__(
         """
         if not path:
             path = "/"
+
+        lilya_permissions = [
+            wrap_permission(permission)
+            for permission in permissions or []
+            if not is_esmerald_permission(permission)
+        ]
+
         super().__init__(
             path=path,
             handler=handler,
             include_in_schema=include_in_schema,
             exception_handlers=exception_handlers,
             name=name,
+            permissions=lilya_permissions,
         )
 
         self._permissions: Union[List[Permission], VoidType] = Void
@@ -2035,7 +2053,9 @@ def __init__(
             description = inspect.cleandoc(self.handler.__doc__ or "") if self.handler else ""
 
         self.description = description
-        self.permissions = list(permissions) if permissions else []  # type: ignore
+        self.permissions = [
+            permission for permission in permissions or [] if is_esmerald_permission(permission)
+        ]  # type: ignore
         self.interceptors: Sequence[Interceptor] = []
         self.middleware = list(middleware) if middleware else []
         self.description = self.description.split("\f")[0]
@@ -2342,6 +2362,7 @@ def __init__(
         _path: str = None
         if not path:
             _path = "/"  # pragma: no cover
+
         super().__init__(
             path=_path,
             handler=handler,
@@ -2407,7 +2428,18 @@ def __init__(
     ):
         if not path:
             path = "/"
-        super().__init__(path=path, handler=handler, exception_handlers=exception_handlers)
+
+        lilya_permissions = [
+            wrap_permission(permission)
+            for permission in permissions or []
+            if not is_esmerald_permission(permission)
+        ]
+        super().__init__(
+            path=path,
+            handler=handler,
+            exception_handlers=exception_handlers,
+            permissions=lilya_permissions,
+        )
         self._permissions: Union[List[Permission], VoidType] = Void
         self._dependencies: Dependencies = {}
         self._response_handler: Union[Callable[[Any], Awaitable[LilyaResponse]], VoidType] = Void
@@ -2416,7 +2448,9 @@ def __init__(
         self.handler = handler
         self.parent: ParentType = None
         self.dependencies = dependencies
-        self.permissions = permissions  # type: ignore
+        self.permissions: Sequence[Permission] = [
+            permission for permission in permissions or [] if is_esmerald_permission(permission)
+        ]  # type: ignore
         self.middleware = middleware
         self.signature_model: Optional[Type[SignatureModel]] = None
         self.websocket_parameter_model: Optional[TransformerModel] = None
@@ -2891,6 +2925,11 @@ async def another(request: Request) -> str:
         if routes:
             routes = self.resolve_route_path_handler(routes)
 
+        lilya_permissions = [
+            wrap_permission(permission)
+            for permission in permissions or []
+            if not is_esmerald_permission(permission)
+        ]
         super().__init__(
             path=self.path,
             app=self.app,
@@ -2901,10 +2940,13 @@ async def another(request: Request) -> str:
             deprecated=deprecated,
             include_in_schema=include_in_schema,
             redirect_slashes=redirect_slashes,
+            permissions=lilya_permissions,
         )
 
         # Making sure Esmerald uses the Esmerald permission system and not Lilya's.
-        self.permissions: Sequence[Permission] = permissions or []  # type: ignore
+        self.permissions: Sequence[Permission] = [
+            permission for permission in permissions or [] if is_esmerald_permission(permission)
+        ]  # type: ignore
 
     def resolve_app_parent(self, app: Optional[Any]) -> Optional[Any]:
         """

tests/permissions/test_permissions.py

@@ -1,11 +1,13 @@
 from typing import TYPE_CHECKING
 
 import pytest
+from lilya.protocols.permissions import PermissionProtocol
 from lilya.status import HTTP_200_OK, HTTP_403_FORBIDDEN
 from lilya.websockets import WebSocketDisconnect
 
 from esmerald.applications import ChildEsmerald
 from esmerald.permissions import AllowAny, BasePermission, DenyAll
+from esmerald.permissions.utils import is_esmerald_permission
 from esmerald.requests import Request
 from esmerald.routing.gateways import Gateway, WebSocketGateway
 from esmerald.routing.handlers import get, route, websocket
@@ -153,3 +155,29 @@ async def my_asgi_handler() -> None:
         assert (
             response.json().get("detail") == "You do not have permission to perform this action."
         )
+
+
+class TestPermission:
+    def has_permission(self, request: "Request", apiview: "APIGateHandler"):
+        return False
+
+
+class DummyPermission(PermissionProtocol):
+    def has_permission(self, request: "Request", apiview: "APIGateHandler"):
+        return False
+
+
+class DummyTrue(BasePermission): ...
+
+
+@pytest.mark.parametrize(
+    "permission,result",
+    [
+        (AllowAny, True),
+        (DummyTrue, True),
+        (TestPermission, False),
+        (DummyPermission, False),
+    ],
+)
+def test_is_esmerald_permission(permission, result) -> None:
+    assert is_esmerald_permission(permission) == result