chore: reduce number of mypy warnings

integration_tests/subroutes/__init__.py

@@ -1,4 +1,4 @@
-from robyn import SubRouter, WebSocket, jsonify
+from robyn import SubRouter, WebSocket, jsonify, Request
 
 from .di_subrouter import di_subrouter
 
@@ -68,3 +68,20 @@ def head_foo():
 def sample_subrouter_openapi_endpoint():
     """Get subrouter openapi"""
     return 200
+
+
+# ===== Authentication =====
+
+
+@sub_router.get("/sync/auth", auth_required=True)
+def sync_auth(request: Request):
+    assert request.identity is not None
+    assert request.identity.claims == {"key": "value"}
+    return "authenticated"
+
+
+@sub_router.get("/async/auth", auth_required=True)
+async def async_auth(request: Request):
+    assert request.identity is not None
+    assert request.identity.claims == {"key": "value"}
+    return "authenticated"

integration_tests/subroutes/di_subrouter.py

@@ -15,3 +15,20 @@ def sync_subrouter_route_dependency(r: Request, router_dependencies, global_depe
 @di_subrouter.get("/subrouter_global_di")
 def sync_subrouter_global_dependency(global_dependencies):
     return global_dependencies["GLOBAL_DEPENDENCY"]
+
+
+# ===== Authentication =====
+
+
+@di_subrouter.get("/subrouter_router_di/sync/auth", auth_required=True)
+def sync_subrouter_auth(request: Request):
+    assert request.identity is not None
+    assert request.identity.claims == {"key": "value"}
+    return "authenticated"
+
+
+@di_subrouter.get("/subrouter_router_di/async/auth", auth_required=True)
+async def async_subrouter_auth(request: Request):
+    assert request.identity is not None
+    assert request.identity.claims == {"key": "value"}
+    return "authenticated"

integration_tests/test_subrouter_authentication.py

@@ -0,0 +1,42 @@
+import pytest
+
+from integration_tests.helpers.http_methods_helpers import get
+
+
+@pytest.mark.benchmark
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_valid_subroute_authentication(session, function_type: str):
+    r = get(f"/sub_router/{function_type}/auth", headers={"Authorization": "Bearer valid"})
+    assert r.text == "authenticated"
+
+
+@pytest.mark.benchmark
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_invalid_subroute_authentication_token(session, function_type: str):
+    r = get(
+        f"/sub_router/{function_type}/auth",
+        headers={"Authorization": "Bearer invalid"},
+        should_check_response=False,
+    )
+    assert r.status_code == 401
+    assert r.headers.get("WWW-Authenticate") == "BearerGetter"
+
+
+@pytest.mark.benchmark
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_invalid_subroute_authentication_header(session, function_type: str):
+    r = get(
+        f"/sub_router/{function_type}/auth",
+        headers={"Authorization": "Bear valid"},
+        should_check_response=False,
+    )
+    assert r.status_code == 401
+    assert r.headers.get("WWW-Authenticate") == "BearerGetter"
+
+
+@pytest.mark.benchmark
+@pytest.mark.parametrize("function_type", ["sync", "async"])
+def test_invalid_subroute_authentication_no_token(session, function_type: str):
+    r = get(f"/sub_router/{function_type}/auth", should_check_response=False)
+    assert r.status_code == 401
+    assert r.headers.get("WWW-Authenticate") == "BearerGetter"

robyn/__init__.py

@@ -5,8 +5,8 @@
 from pathlib import Path
 from typing import Callable, List, Optional, Tuple, Union
 
-import multiprocess as mp
-from nestd import get_all_nested
+import multiprocess as mp  # type: ignore
+from nestd import get_all_nested  # type: ignore
 
 from robyn import status_codes
 from robyn.argument_parser import Config
@@ -42,7 +42,7 @@ def __init__(
         self,
         file_object: str,
         config: Config = Config(),
-        openapi_file_path: str = None,
+        openapi_file_path: Optional[str] = None,
         openapi: OpenAPI = OpenAPI(),
         dependencies: DependencyMap = DependencyMap(),
     ) -> None:
@@ -79,7 +79,7 @@ def __init__(
         self.response_headers: Headers = Headers({})
         self.excluded_response_headers_paths: Optional[List[str]] = None
         self.directories: List[Directory] = []
-        self.event_handlers = {}
+        self.event_handlers: dict = {}
         self.exception_handler: Optional[Callable] = None
         self.authentication_handler: Optional[AuthenticationHandler] = None
 
@@ -596,29 +596,29 @@ def __init__(self, file_object: str, prefix: str = "", config: Config = Config()
     def __add_prefix(self, endpoint: str):
         return f"{self.prefix}{endpoint}"
 
-    def get(self, endpoint: str, const: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["get"]):
-        return super().get(endpoint=self.__add_prefix(endpoint), const=const, openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def get(self, endpoint: str, const: bool = False, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["get"]):
+        return super().get(endpoint=self.__add_prefix(endpoint), const=const, auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
-    def post(self, endpoint: str, openapi_name: str = "", openapi_tags: List[str] = ["post"]):
-        return super().post(endpoint=self.__add_prefix(endpoint), openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def post(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["post"]):
+        return super().post(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
-    def put(self, endpoint: str, openapi_name: str = "", openapi_tags: List[str] = ["put"]):
-        return super().put(endpoint=self.__add_prefix(endpoint), openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def put(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["put"]):
+        return super().put(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
-    def delete(self, endpoint: str, openapi_name: str = "", openapi_tags: List[str] = ["delete"]):
-        return super().delete(endpoint=self.__add_prefix(endpoint), openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def delete(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["delete"]):
+        return super().delete(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
-    def patch(self, endpoint: str, openapi_name: str = "", openapi_tags: List[str] = ["patch"]):
-        return super().patch(endpoint=self.__add_prefix(endpoint), openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def patch(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["patch"]):
+        return super().patch(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
-    def head(self, endpoint: str, openapi_name: str = "", openapi_tags: List[str] = ["head"]):
-        return super().head(endpoint=self.__add_prefix(endpoint), openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def head(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["head"]):
+        return super().head(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
-    def trace(self, endpoint: str, openapi_name: str = "", openapi_tags: List[str] = ["trace"]):
-        return super().trace(endpoint=self.__add_prefix(endpoint), openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def trace(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["trace"]):
+        return super().trace(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
-    def options(self, endpoint: str, openapi_name: str = "", openapi_tags: List[str] = ["options"]):
-        return super().options(endpoint=self.__add_prefix(endpoint), openapi_name=openapi_name, openapi_tags=openapi_tags)
+    def options(self, endpoint: str, auth_required: bool = False, openapi_name: str = "", openapi_tags: List[str] = ["options"]):
+        return super().options(endpoint=self.__add_prefix(endpoint), auth_required=auth_required, openapi_name=openapi_name, openapi_tags=openapi_tags)
 
 
 def ALLOW_CORS(app: Robyn, origins: Union[List[str], str]):

robyn/argument_parser.py

@@ -103,7 +103,8 @@ def __init__(self) -> None:
         if self.fast:
             # doing this here before every other check
             # so that processes, workers and log_level can be overridden
-            self.processes = self.processes or ((os.cpu_count() * 2) + 1) or 1
+            cpu_count: int = os.cpu_count() or 1
+            self.processes = self.processes or ((cpu_count * 2) + 1) or 1
             self.workers = self.workers or 2
             self.log_level = self.log_level or "WARNING"
 

robyn/authentication.py

@@ -81,7 +81,7 @@ class BearerGetter(TokenGetter):
 
     @classmethod
     def get_token(cls, request: Request) -> Optional[str]:
-        if "authorization" in request.headers:
+        if request.headers.contains("authorization"):
             authorization_header = request.headers.get("authorization")
         else:
             authorization_header = None

robyn/dependency_injection.py

@@ -4,7 +4,7 @@
 
 
 class DependencyMap:
-    def __init__(self):
+    def __init__(self) -> None:
         self.global_dependency_map: dict[str, Any] = {}
         # {'router': {'dependency_name': dependency_class}
         self.router_dependency_map: dict[str, dict[str, Any]] = {}

robyn/openapi.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, TypedDict
 
-from robyn.responses import FileResponse, html
+from robyn.responses import html
 from robyn.robyn import QueryParams, Response
 from robyn.types import Body
 
@@ -239,13 +239,13 @@ def get_path_obj(
         query_params: Optional[TypedDict],
         request_body: Optional[TypedDict],
         return_annotation: Optional[TypedDict],
-    ) -> (str, dict):
+    ) -> typing.Tuple[str, dict]:
         """
         Get the "path" openapi object according to spec
 
         @param endpoint: str the endpoint to be added
         @param name: str the name of the endpoint
-        @param description: Optional[str] short description of the endpoint (to be fetched from the endpoint defenition by default)
+        @param description: Optional[str] short description of the endpoint (to be fetched from the endpoint definition by default)
         @param tags: List[str] for grouping of endpoints
         @param query_params: Optional[TypedDict] query params for the function
         @param request_body: Optional[TypedDict] request body for the function
@@ -328,7 +328,7 @@ def get_path_obj(
 
             openapi_path_object["requestBody"] = request_body_object
 
-        response_schema = {}
+        response_schema: dict = {}
         response_type = "text/plain"
 
         if return_annotation and return_annotation is not Response:
@@ -416,7 +416,7 @@ def override_openapi(self, openapi_json_spec_path: Path):
             self.openapi_spec = dict(json_file_content)
             self.openapi_file_override = True
 
-    def get_openapi_docs_page(self) -> FileResponse:
+    def get_openapi_docs_page(self) -> Response:
         """
         Handler to the swagger html page to be deployed to the endpoint `/docs`
         @return: FileResponse the swagger html page

robyn/processpool.py

@@ -4,7 +4,7 @@
 import webbrowser
 from typing import Dict, List, Optional
 
-from multiprocess import Process
+from multiprocess import Process  # type: ignore
 
 from robyn.events import Events
 from robyn.logger import logger
@@ -80,7 +80,7 @@ def init_processpool(
     response_headers: Headers,
     excluded_response_headers_paths: Optional[List[str]],
 ) -> List[Process]:
-    process_pool = []
+    process_pool: List = []
     if sys.platform.startswith("win32") or processes == 1:
         spawn_process(
             directories,

robyn/reloader.py

@@ -118,7 +118,7 @@ def __init__(self, file_path: str, directory_path: str) -> None:
         self.file_path = file_path
         self.directory_path = directory_path
         self.process = None  # Keep track of the subprocess
-        self.built_rust_binaries = []  # Keep track of the built rust binaries
+        self.built_rust_binaries: List = []  # Keep track of the built rust binaries
 
         self.last_reload = time.time()  # Keep track of the last reload. EventHandler is initialized with the process.