🚨 Fix mypy errors

Author: rushilsrivastava

fastapi_gcp_tasks/decorators.py

@@ -1,8 +1,13 @@
-def task_default_options(**kwargs):
+from typing import Any, Callable, TypeVar
+
+F = TypeVar("F", bound=Callable[..., Any])
+
+
+def task_default_options(**kwargs: Any) -> Callable[[F], F]:
     """Wrapper to set default options for a cloud task."""
 
-    def wrapper(fn):
-        fn._delay_options = kwargs
+    def wrapper(fn: F) -> F:
+        fn._delay_options = kwargs  # type: ignore[attr-defined]
         return fn
 
     return wrapper

fastapi_gcp_tasks/delayed_route.py

@@ -1,5 +1,5 @@
 # Standard Library Imports
-from typing import Callable
+from typing import Callable, Type
 
 # Third Party Imports
 from fastapi.routing import APIRoute
@@ -16,10 +16,10 @@ def DelayedRouteBuilder(  # noqa: N802
     base_url: str,
     queue_path: str,
     task_create_timeout: float = 10.0,
-    pre_create_hook: DelayedTaskHook = None,
-    client=None,
-    auto_create_queue=True,
-):
+    pre_create_hook: DelayedTaskHook | None = None,
+    client: tasks_v2.CloudTasksClient | None = None,
+    auto_create_queue: bool = True,
+) -> Type[APIRoute]:
     """
     Returns a Mixin that should be used to override route_class.
 
@@ -57,11 +57,11 @@ def on_user_create(user_id: str, data: UserData):
     class TaskRouteMixin(APIRoute):
         def get_route_handler(self) -> Callable:
             original_route_handler = super().get_route_handler()
-            self.endpoint.options = self.delay_options
-            self.endpoint.delay = self.delay
+            self.endpoint.options = self.delay_options  # type: ignore[attr-defined]
+            self.endpoint.delay = self.delay  # type: ignore[attr-defined]
             return original_route_handler
 
-        def delay_options(self, **options) -> Delayer:
+        def delay_options(self, **options: dict) -> Delayer:
             delay_opts = {
                 "base_url": base_url,
                 "queue_path": queue_path,
@@ -73,12 +73,13 @@ def delay_options(self, **options) -> Delayer:
                 delay_opts |= self.endpoint._delay_options
             delay_opts |= options
 
+            # ignoring the type here because the dictionary values are unpacked
             return Delayer(
                 route=self,
-                **delay_opts,
+                **delay_opts,  # type: ignore[arg-type]
             )
 
-        def delay(self, **kwargs):
+        def delay(self, **kwargs: dict) -> tasks_v2.Task:
             return self.delay_options().delay(**kwargs)
 
     return TaskRouteMixin

fastapi_gcp_tasks/delayer.py

@@ -1,5 +1,6 @@
 # Standard Library Imports
 import datetime
+from typing import Any, Iterable
 
 # Third Party Imports
 from fastapi.routing import APIRoute
@@ -38,7 +39,7 @@ def __init__(
         pre_create_hook: DelayedTaskHook,
         task_create_timeout: float = 10.0,
         countdown: int = 0,
-        task_id: str = None,
+        task_id: str | None = None,
     ) -> None:
         super().__init__(route=route, base_url=base_url)
         self.queue_path = queue_path
@@ -50,7 +51,7 @@ def __init__(
         self.client = client
         self.pre_create_hook = pre_create_hook
 
-    def delay(self, **kwargs):
+    def delay(self, **kwargs: Any) -> tasks_v2.Task:
         """Delay a task on Cloud Tasks."""
         # Create http request
         request = tasks_v2.HttpRequest()
@@ -76,7 +77,7 @@ def delay(self, **kwargs):
 
         return self.client.create_task(request=request, timeout=self.task_create_timeout)
 
-    def _schedule(self):
+    def _schedule(self) -> timestamp_pb2.Timestamp | None:
         if self.countdown is None or self.countdown <= 0:
             return None
         d = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=self.countdown)
@@ -85,7 +86,7 @@ def _schedule(self):
         return timestamp
 
 
-def _task_method(methods):
+def _task_method(methods: Iterable[str]) -> tasks_v2.HttpMethod:
     method_map = {
         "POST": tasks_v2.HttpMethod.POST,
         "GET": tasks_v2.HttpMethod.GET,
@@ -102,4 +103,4 @@ def _task_method(methods):
     method = method_map.get(methods[0])
     if method is None:
         raise BadMethodError(f"Unknown method {methods[0]}")
-    return method
+    return tasks_v2.HttpMethod(method)

fastapi_gcp_tasks/dependencies.py

@@ -1,18 +1,19 @@
 # Standard Library Imports
-import typing
 from datetime import datetime
+from typing import Any, Callable
 
 # Third Party Imports
 from fastapi import Depends, Header, HTTPException
 
 
-def max_retries(count: int = 20):
-    """Raises a http exception (with status 200) after max retries are exhausted."""
+def max_retries(count: int = 20) -> Callable[[Any], bool]:
+    """Raises an http exception (with status 200) after max retries are exhausted."""
 
     def retries_dep(meta: CloudTasksHeaders = Depends()) -> bool:
         # count starts from 0 so equality check is required
         if meta.retry_count >= count:
             raise HTTPException(status_code=200, detail="Max retries exhausted")
+        return True
 
     return retries_dep
 
@@ -26,13 +27,13 @@ class CloudTasksHeaders:
 
     def __init__(
         self,
-        x_cloudtasks_taskretrycount: typing.Optional[int] = Header(0),
-        x_cloudtasks_taskexecutioncount: typing.Optional[int] = Header(0),
-        x_cloudtasks_queuename: typing.Optional[str] = Header(""),
-        x_cloudtasks_taskname: typing.Optional[str] = Header(""),
-        x_cloudtasks_tasketa: typing.Optional[float] = Header(0),
-        x_cloudtasks_taskpreviousresponse: typing.Optional[int] = Header(0),
-        x_cloudtasks_taskretryreason: typing.Optional[str] = Header(""),
+        x_cloudtasks_taskretrycount: int = Header(0),
+        x_cloudtasks_taskexecutioncount: int = Header(0),
+        x_cloudtasks_queuename: str = Header(""),
+        x_cloudtasks_taskname: str = Header(""),
+        x_cloudtasks_tasketa: float = Header(0),
+        x_cloudtasks_taskpreviousresponse: int = Header(0),
+        x_cloudtasks_taskretryreason: str = Header(""),
     ) -> None:
         self.retry_count = x_cloudtasks_taskretrycount
         self.execution_count = x_cloudtasks_taskexecutioncount

fastapi_gcp_tasks/hooks.py

@@ -1,5 +1,5 @@
 # Standard Library Imports
-from typing import Callable
+from typing import Any, Callable
 
 # Third Party Imports
 from google.cloud import scheduler_v1, tasks_v2
@@ -9,15 +9,15 @@
 ScheduledHook = Callable[[scheduler_v1.CreateJobRequest], scheduler_v1.CreateJobRequest]
 
 
-def noop_hook(request):
+def noop_hook(request: Any) -> Any:
     """Inspired by https://github.com/kelseyhightower/nocode."""
     return request
 
 
-def chained_hook(*hooks):
+def chained_hook(*hooks: Callable[[Any], Any]) -> Callable[[Any], Any]:
     """Call all hooks sequentially with the result from the previous hook."""
 
-    def chain(request):
+    def chain(request: Any) -> Any:
         for hook in hooks:
             request = hook(request)
         return request

fastapi_gcp_tasks/requester.py

@@ -1,5 +1,5 @@
 # Standard Library Imports
-from typing import Dict, List, Tuple
+from typing import Any, Dict, List, Tuple
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
 # Third Party Imports
@@ -16,7 +16,7 @@
     import ujson as json
 except ImportError:
     # Standard Library Imports
-    import json
+    import json  # type: ignore[no-redef]
 
 
 class Requester:
@@ -39,7 +39,7 @@ def __init__(
         self.route = route
         self.base_url = base_url.rstrip("/")
 
-    def _headers(self, *, values):
+    def _headers(self, *, values: Dict[str, Any]) -> Dict[str, str]:
         headers = _err_val(request_params_to_args(self.route.dependant.header_params, values))
         cookies = _err_val(request_params_to_args(self.route.dependant.cookie_params, values))
         if len(cookies) > 0:
@@ -49,7 +49,7 @@ def _headers(self, *, values):
         # Always send string headers and skip all headers which are supposed to be sent by cloudtasks
         return {str(k): str(v) for (k, v) in headers.items() if not str(k).startswith("x_cloudtasks_")}
 
-    def _url(self, *, values):
+    def _url(self, *, values: Dict[str, Any]) -> str:
         route = self.route
         path_values = _err_val(request_params_to_args(route.dependant.path_params, values))
         for name, converter in route.param_convertors.items():
@@ -80,11 +80,11 @@ def _url(self, *, values):
         url_parts[4] = urlencode(query)
         return urlunparse(url_parts)
 
-    def _body(self, *, values):
+    def _body(self, *, values: Dict[str, Any]) -> bytes | None:
         body = None
         body_field = self.route.body_field
         if body_field and body_field.name:
-            got_body = values.get(body_field.name, None)
+            got_body = values.get(body_field.name)
             if got_body is None:
                 if body_field.required:
                     raise MissingParamError(name=body_field.name)
@@ -95,7 +95,7 @@ def _body(self, *, values):
         return body
 
 
-def _err_val(resp: Tuple[Dict, List[ErrorWrapper]]):
+def _err_val(resp: Tuple[Dict, List[ErrorWrapper]]) -> Dict:
     values, errors = resp
 
     if len(errors) != 0:

fastapi_gcp_tasks/scheduled_route.py

@@ -1,5 +1,5 @@
 # Standard Library Imports
-from typing import Callable
+from typing import Callable, Type
 
 # Third Party Imports
 from fastapi.routing import APIRoute
@@ -15,9 +15,9 @@ def ScheduledRouteBuilder(  # noqa: N802
     base_url: str,
     location_path: str,
     job_create_timeout: float = 10.0,
-    pre_create_hook: ScheduledHook = None,
-    client=None,
-):
+    pre_create_hook: ScheduledHook | None = None,
+    client: scheduler_v1.CloudSchedulerClient | None = None,
+) -> Type[APIRoute]:
     """
     Returns a Mixin that should be used to override route_class.
 
@@ -47,10 +47,10 @@ def simple_scheduled_task():
     class ScheduledRouteMixin(APIRoute):
         def get_route_handler(self) -> Callable:
             original_route_handler = super().get_route_handler()
-            self.endpoint.scheduler = self.scheduler_options
+            self.endpoint.scheduler = self.scheduler_options  # type: ignore[attr-defined]
             return original_route_handler
 
-        def scheduler_options(self, *, name, schedule, **options) -> Scheduler:
+        def scheduler_options(self, *, name: str, schedule: str, **options: dict) -> Scheduler:
             scheduler_opts = {
                 "base_url": base_url,
                 "location_path": location_path,
@@ -60,6 +60,8 @@ def scheduler_options(self, *, name, schedule, **options) -> Scheduler:
                 "name": name,
                 "schedule": schedule,
             } | options
-            return Scheduler(route=self, **scheduler_opts)
+
+            # ignoring the type here because the dictionary values are unpacked
+            return Scheduler(route=self, **scheduler_opts)  # type: ignore[arg-type]
 
     return ScheduledRouteMixin

fastapi_gcp_tasks/scheduler.py

@@ -1,4 +1,5 @@
 # Standard Library Imports
+from typing import Any, Iterable
 
 # Third Party Imports
 from fastapi.routing import APIRoute
@@ -41,7 +42,7 @@ def __init__(
         pre_create_hook: ScheduledHook,
         name: str = "",
         job_create_timeout: float = 10.0,
-        retry_config: scheduler_v1.RetryConfig = None,
+        retry_config: scheduler_v1.RetryConfig | None = None,
         time_zone: str = "UTC",
         force: bool = False,
     ) -> None:
@@ -73,7 +74,7 @@ def __init__(
         self.pre_create_hook = pre_create_hook
         self.force = force
 
-    def schedule(self, **kwargs):
+    def schedule(self, **kwargs: Any) -> None:
         """Schedule a job on Cloud Scheduler."""
         # Create http request
         request = scheduler_v1.HttpTarget()
@@ -103,15 +104,15 @@ def schedule(self, **kwargs):
             self.delete()
             self.client.create_job(request=request, timeout=self.job_create_timeout)
 
-    def _has_changed(self, request: scheduler_v1.CreateJobRequest):
+    def _has_changed(self, request: scheduler_v1.CreateJobRequest) -> bool:
         try:
             job = self.client.get_job(name=request.job.name)
             # Remove things that are either output only or GCP adds by default
-            job.user_update_time = None
-            job.state = None
+            job.user_update_time = None  # type: ignore[assignment]
+            job.state = None  # type: ignore[assignment]
             job.status = None
-            job.last_attempt_time = None
-            job.schedule_time = None
+            job.last_attempt_time = None  # type: ignore[assignment]
+            job.schedule_time = None  # type: ignore[assignment]
             del job.http_target.headers["User-Agent"]
             # Proto compare works directly with `__eq__`
             return job != request.job
@@ -120,7 +121,7 @@ def _has_changed(self, request: scheduler_v1.CreateJobRequest):
             return True
         return False
 
-    def delete(self):
+    def delete(self) -> bool | Exception:
         """Delete the job from the scheduler if it exists."""
         # We return true or exception because you could have the delete code on multiple instances
         try:
@@ -131,7 +132,7 @@ def delete(self):
             return ex
 
 
-def _scheduler_method(methods):
+def _scheduler_method(methods: Iterable[str]) -> scheduler_v1.HttpMethod:
     method_map = {
         "POST": scheduler_v1.HttpMethod.POST,
         "GET": scheduler_v1.HttpMethod.GET,
@@ -148,4 +149,4 @@ def _scheduler_method(methods):
     method = method_map.get(methods[0])
     if method is None:
         raise BadMethodError(f"Unknown method {methods[0]}")
-    return method
+    return scheduler_v1.HttpMethod(method)

fastapi_gcp_tasks/utils.py

@@ -1,16 +1,18 @@
 # Third Party Imports
+from typing import Any
+
 import grpc
 from google.api_core.exceptions import AlreadyExists
 from google.cloud import scheduler_v1, tasks_v2
 from google.cloud.tasks_v2.services.cloud_tasks import transports
 
 
-def location_path(*, project: str, location: str, **ignored):
+def location_path(*, project: str, location: str) -> str:
     """Helper function to construct a location path for Cloud Scheduler."""
     return scheduler_v1.CloudSchedulerClient.common_location_path(project=project, location=location)
 
 
-def queue_path(*, project: str, location: str, queue: str):
+def queue_path(*, project: str, location: str, queue: str) -> str:
     """Helper function to construct a queue path for Cloud Tasks."""
     return tasks_v2.CloudTasksClient.queue_path(project=project, location=location, queue=queue)
 
@@ -19,8 +21,8 @@ def ensure_queue(
     *,
     client: tasks_v2.CloudTasksClient,
     path: str,
-    **kwargs,
-):
+    **kwargs: Any,
+) -> None:
     """
     Helper function to ensure a Cloud Tasks queue exists.
 
@@ -39,7 +41,7 @@ def ensure_queue(
         pass
 
 
-def emulator_client(*, host="localhost:8123"):
+def emulator_client(*, host: str = "localhost:8123") -> tasks_v2.CloudTasksClient:
     """Helper function to create a CloudTasksClient from an emulator host."""
     channel = grpc.insecure_channel(host)
     transport = transports.CloudTasksGrpcTransport(channel=channel)