Initiate type hinting for improved code clarity and consistency

pyblaze/cli/generate_repository_file.py

@@ -53,7 +53,9 @@ def __init__(self, {singularized_module_name}_repository: {module_name_capitaliz
     )
 
     # Update the regular expression to match @path or @websocket
-    class_pattern = re.compile(r"@(path|websocket)\(.+\)\nclass\s+[A-Za-z_][A-Za-z0-9_]*:")
+    class_pattern = re.compile(
+        r"@(path|websocket)\(.+\)\nclass\s+[A-Za-z_][A-Za-z0-9_]*:"
+    )
     match = class_pattern.search(existing_content)
 
     if match:

pyblaze/decorators/http_methods.py

@@ -1,8 +1,11 @@
 from pyblaze.enums import HttpMethod
 
 
-def get(path: str):
-    def decorator(handler):
+from typing import Callable, Type
+
+
+def get(path: str) -> Callable[[Type], Type]:
+    def decorator(handler: Type) -> Type:
         handler.__method__ = HttpMethod.GET
         handler.__path__ = path
         handler.__is_handler__ = True
@@ -11,8 +14,8 @@ def decorator(handler):
     return decorator
 
 
-def post(path: str):
-    def decorator(handler):
+def post(path: str) -> Callable[[Type], Type]:
+    def decorator(handler: Type) -> Type:
         handler.__method__ = HttpMethod.POST
         handler.__path__ = path
         handler.__is_handler__ = True
@@ -21,8 +24,8 @@ def decorator(handler):
     return decorator
 
 
-def put(path: str):
-    def decorator(handler):
+def put(path: str) -> Callable[[Type], Type]:
+    def decorator(handler: Type) -> Type:
         handler.__method__ = HttpMethod.PUT
         handler.__path__ = path
         handler.__is_handler__ = True
@@ -31,8 +34,8 @@ def decorator(handler):
     return decorator
 
 
-def delete(path: str):
-    def decorator(handler):
+def delete(path: str) -> Callable[[Type], Type]:
+    def decorator(handler: Type) -> Type:
         handler.__method__ = HttpMethod.DELETE
         handler.__path__ = path
         handler.__is_handler__ = True

pyblaze/decorators/path.py

@@ -1,5 +1,8 @@
-def path(path: str):
-    def decorator(cls):
+from typing import Type, Callable
+
+
+def path(path: str) -> Callable[[Type], Type]:
+    def decorator(cls: Type) -> Type:
         cls.__path__ = path
         cls.__is_controller__ = True
         return cls

pyblaze/decorators/websocket.py

@@ -1,7 +1,7 @@
 from pyblaze.websockets import WebSocketControllerRegistry
 
 
-def websocket(path):
+def websocket(path: str):
     def decorator(cls):
         WebSocketControllerRegistry.register_controller(path, cls)
         return cls

pyblaze/helpers.py

@@ -50,7 +50,7 @@
 """
 
 
-def format_server_exception():
+def format_server_exception() -> HttpResponse:
     msg = template.format(
         message="Internal Server Error"
         "<br><br>We are currently trying to fix the problem.",
@@ -59,7 +59,7 @@ def format_server_exception():
     return HttpResponse(content=msg, status_code=500, content_type="text/html")
 
 
-def format_not_found_exception():
+def format_not_found_exception() -> HttpResponse:
     msg = template.format(
         message="Ooops!!! The page you are looking for is not found", status_code=404
     )

pyblaze/pyblaze.py

@@ -5,7 +5,7 @@
 import os
 import re
 import sys
-from typing import Callable, Dict, List
+from typing import Callable, Dict, List, Any
 
 from httpx import AsyncClient
 
@@ -29,19 +29,19 @@ def __init__(self, secret_key=None, session_type=None):
         self.session_type = session_type
         self.discover_controllers()
 
-    def add_middleware(self, middleware: Callable):
+    def add_middleware(self, middleware: Callable) -> Callable:
         self.middleware.append(middleware)
         return middleware
 
-    def add_route(self, path: str, method: HttpMethod, handler: Callable):
+    def add_route(self, path: str, method: HttpMethod, handler: Callable) -> None:
         if path in self.routes[method.value]:
             raise AssertionError(
                 f"Route with method '{method}' and path '{path}' already exists"
             )
 
         self.routes[method.value][path] = handler
 
-    def discover_controllers(self):
+    def discover_controllers(self) -> None:
         current_dir = os.getcwd()
         src_dir = os.path.join(current_dir, "src")
 
@@ -55,7 +55,7 @@ def discover_controllers(self):
                     module_path = f"src.{rel_path}"
                     self._add_routes(module_path)
 
-    def _add_routes(self, file_path: str):
+    def _add_routes(self, file_path: str) -> None:
         try:
             module_name = self._file_path_to_module(file_path)
             src_directory = os.path.abspath(
@@ -72,7 +72,7 @@ def _add_routes(self, file_path: str):
             # Remove the 'src' directory from the Python path after importing
             sys.path.pop(0)
 
-    def _add_class_routes(self, cls):
+    def _add_class_routes(self, cls) -> None:
         if not hasattr(cls, "__path__"):
             return
 
@@ -92,11 +92,11 @@ def _add_class_routes(self, cls):
                 full_route = path_prefix + route
                 self.add_route(full_route, http_method, method)
 
-    def _file_path_to_module(self, file_path: str):
+    def _file_path_to_module(self, file_path: str) -> str:
         rel_path = os.path.relpath(file_path, os.getcwd())
         return rel_path.replace(os.sep, ".")
 
-    def _parse_controller_decorators(self, file_path):
+    def _parse_controller_decorators(self, file_path: str) -> bool:
         with open(file_path, "r") as file:
             tree = ast.parse(file.read(), filename=file_path)
 
@@ -114,18 +114,22 @@ def _parse_controller_decorators(self, file_path):
                         return True
         return False
 
-    def _is_controller_file(self, file_path):
+    def _is_controller_file(self, file_path: str) -> bool:
         return file_path.endswith(
             "_controller.py"
         ) and self._parse_controller_decorators(file_path)
 
-    async def __call__(self, scope, receive, send):
+    async def __call__(
+        self, scope: Dict[str, Any], receive: Callable, send: Callable
+    ) -> None:
         if scope["type"] == "websocket":
             await handle_websocket(scope, receive, send)
         elif scope["type"] == "http":
-            await self.handle_http(receive, scope, send)
+            await self.handle_http(scope, receive, send)
 
-    async def handle_http(self, receive, scope, send):
+    async def handle_http(
+        self, scope: Dict[str, Any], receive: Callable, send: Callable
+    ) -> None:
         request = await self.create_request(receive, scope)
         RequestContext.set_request(request)
 
@@ -142,7 +146,15 @@ async def handle_http(self, receive, scope, send):
         else:
             await self.handle_dynamic_route(method, path, request, scope, receive, send)
 
-    async def handle_dynamic_route(self, method, path, request, scope, receive, send):
+    async def handle_dynamic_route(
+        self,
+        method: str,
+        path: str,
+        request: Request,
+        scope: Dict[str, Any],
+        receive: Callable,
+        send: Callable,
+    ):
         for route_path, handler in self.routes[method].items():
             if "{" in route_path and "}" in route_path:
                 route_pattern = route_path.replace("{", "(?P<").replace("}", ">[^/]+)")
@@ -161,7 +173,15 @@ async def handle_dynamic_route(self, method, path, request, scope, receive, send
         # If no matching route is found, return a 404 response
         await self.handle_not_found(scope, receive, send)
 
-    async def handle_route(self, method, path, receive, request, scope, send):
+    async def handle_route(
+        self,
+        method: str,
+        path: str,
+        receive: Callable,
+        request: Request,
+        scope: Dict[str, Any],
+        send: Callable,
+    ):
         try:
             handler = self.routes[method][path]
             response = await self.invoke_handler(handler, request, scope)
@@ -179,27 +199,33 @@ async def handle_route(self, method, path, receive, request, scope, send):
             error_response = await self.error_handler(exc)
             await error_response(scope, receive, send)
 
-    async def process_middlewares(self, request):
+    async def process_middlewares(self, request: Request):
         for middleware in self.middleware:
             request = await middleware(request)
 
-    async def create_request(self, receive, scope):
+    async def create_request(self, receive: Callable, scope: Dict[str, Any]) -> Request:
         return Request(scope, receive)
 
-    async def set_request_session(self, request):
+    async def set_request_session(self, request: Request) -> None:
         if self.session_type:
             session = get_or_create_session(request, self.secret_key)
             request.session = session
 
-    async def _handle_static_files(self, scope, receive, send, request):
+    async def _handle_static_files(
+        self, scope: Dict[str, Any], receive: Callable, send: Callable, request: Request
+    ) -> None:
         template_response = TemplateResponse(request, scope["path"])
         await template_response(scope, receive, send)
 
-    async def handle_not_found(self, scope, receive, send):
+    async def handle_not_found(
+        self, scope: Dict[str, Any], receive: Callable, send: Callable
+    ) -> None:
         not_found_response = format_not_found_exception()
         await not_found_response(scope, receive, send)
 
-    async def invoke_handler(self, handler, request, scope, params=None):
+    async def invoke_handler(
+        self, handler, request: Request, scope: Dict[str, Any], params=None
+    ):
         handler_signature = inspect.signature(handler)
         handler_params = {}
         for param_name, param in handler_signature.parameters.items():

pyblaze/utils/dependency_resolver.py

@@ -1,8 +1,9 @@
 import inspect
+from typing import Type, List, Any, Optional, Callable
 
 
-def resolve_dependencies_automatic(cls):
-    dependencies = []
+def resolve_dependencies_automatic(cls: Type) -> Optional[List[Any]]:
+    dependencies: List[Any] = []
 
     # Get the constructor parameters, if available
     init_method = getattr(cls, "__init__", None)
@@ -18,5 +19,5 @@ def resolve_dependencies_automatic(cls):
     return dependencies if dependencies else None
 
 
-def resolve_dependency(dependency_type):
+def resolve_dependency(dependency_type: Type[Callable]) -> Optional[Callable]:
     return dependency_type() if dependency_type else None

pyblaze/utils/inflection.py

@@ -1,12 +1,14 @@
+from typing import Literal
+
 import inflect
 
 
-def singularize(word):
+def singularize(word: str) -> str | Literal[False]:
     p = inflect.engine()
     return p.singular_noun(word) or word
 
 
-def pluralize_word(word):
+def pluralize_word(word: str) -> str:
     p = inflect.engine()
 
     return p.plural(word) or word

pyblaze/utils/naming.py

@@ -1,15 +1,15 @@
 import re
 
 
-def convert_to_snake_case(input_string):
+def convert_to_snake_case(input_string: str) -> str:
     snake_case_string = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", input_string)
     snake_case_string = snake_case_string.replace("-", "_")
     snake_case_string = snake_case_string.lower()
 
     return snake_case_string
 
 
-def convert_to_camel_case(input_string):
+def convert_to_camel_case(input_string: str) -> str:
     space_separated = input_string.replace("_", " ").replace("-", " ")
     capitalized_words = "".join(word.capitalize() for word in space_separated.split())
     camel_case_string = capitalized_words[0] + capitalized_words[1:]

pyblaze/utils/secret_key.py

@@ -1,5 +1,5 @@
 import secrets
 
 
-def get_random_secret_key(length=50):
+def get_random_secret_key(length: int = 50) -> str:
     return secrets.token_urlsafe(length)[:length]

pyblaze/websockets/websocket_controller_registry.py

@@ -1,11 +1,14 @@
+from typing import Dict, Any
+
+
 class WebSocketControllerRegistry:
     _controllers = {}
 
     @classmethod
-    def register_controller(cls, path, controller_cls):
+    def register_controller(cls, path: str, controller_cls: Dict[str, Any]):
         cls._controllers[path] = controller_cls
         return controller_cls
 
     @classmethod
-    def get_controller(cls, path):
+    def get_controller(cls, path: str):
         return cls._controllers.get(path)