improve decorator types

Author: jyecusch

nitric/resources/apis.py

@@ -181,7 +181,7 @@ def _route(self, match: str, opts: Optional[RouteOptions] = None) -> Route:
         self.routes.append(r)
         return r
 
-    def all(self, match: str, opts: Optional[MethodOptions] = None):
+    def all(self, match: str, opts: Optional[MethodOptions] = None) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to HTTP GET requests."""
 
         def decorator(function: HttpHandler) -> None:
@@ -201,78 +201,80 @@ def decorator(function: HttpHandler) -> None:
 
         return decorator
 
-    def methods(self, methods: List[HttpMethod], match: str, opts: Optional[MethodOptions] = None):
+    def methods(
+        self, methods: List[HttpMethod], match: str, opts: Optional[MethodOptions] = None
+    ) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to specific HTTP requests defined by a list of verbs."""
         if opts is None:
             opts = MethodOptions()
 
-        def decorator(function: HttpHandler):
+        def decorator(function: HttpHandler) -> None:
             r = self._route(match)
             r.method(methods, function, opts=opts)
 
         return decorator
 
-    def get(self, match: str, opts: Optional[MethodOptions] = None):
+    def get(self, match: str, opts: Optional[MethodOptions] = None) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to HTTP GET requests."""
         if opts is None:
             opts = MethodOptions()
 
-        def decorator(function: HttpHandler):
+        def decorator(function: HttpHandler) -> None:
             r = self._route(match)
             r.get(function, opts=opts)
 
         return decorator
 
-    def post(self, match: str, opts: Optional[MethodOptions] = None):
+    def post(self, match: str, opts: Optional[MethodOptions] = None) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to HTTP POST requests."""
         if opts is None:
             opts = MethodOptions()
 
-        def decorator(function: HttpHandler):
+        def decorator(function: HttpHandler) -> None:
             r = self._route(match)
             r.post(function, opts=opts)
 
         return decorator
 
-    def delete(self, match: str, opts: Optional[MethodOptions] = None):
+    def delete(self, match: str, opts: Optional[MethodOptions] = None) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to HTTP DELETE requests."""
         if opts is None:
             opts = MethodOptions()
 
-        def decorator(function: HttpHandler):
+        def decorator(function: HttpHandler) -> None:
             r = self._route(match)
             r.delete(function, opts=opts)
 
         return decorator
 
-    def options(self, match: str, opts: Optional[MethodOptions] = None):
+    def options(self, match: str, opts: Optional[MethodOptions] = None) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to HTTP OPTIONS requests."""
         if opts is None:
             opts = MethodOptions()
 
-        def decorator(function: HttpHandler):
+        def decorator(function: HttpHandler) -> None:
             r = self._route(match)
             r.options(function, opts=opts)
 
         return decorator
 
-    def patch(self, match: str, opts: Optional[MethodOptions] = None):
+    def patch(self, match: str, opts: Optional[MethodOptions] = None) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to HTTP PATCH requests."""
         if opts is None:
             opts = MethodOptions()
 
-        def decorator(function: HttpHandler):
+        def decorator(function: HttpHandler) -> None:
             r = self._route(match)
             r.patch(function, opts=opts)
 
         return decorator
 
-    def put(self, match: str, opts: Optional[MethodOptions] = None):
+    def put(self, match: str, opts: Optional[MethodOptions] = None) -> Callable[[HttpHandler], None]:
         """Define an HTTP route which will respond to HTTP PUT requests."""
         if opts is None:
             opts = MethodOptions()
 
-        def decorator(function: HttpHandler):
+        def decorator(function: HttpHandler) -> None:
             r = self._route(match)
             r.put(function, opts=opts)
 
@@ -311,31 +313,31 @@ def __init__(self, api: Api, path: str, opts: RouteOptions):
 
     def method(
         self, methods: List[HttpMethod], *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None
-    ):
+    ) -> None:
         """Register middleware for multiple HTTP Methods."""
         return Method(self, methods, *middleware, opts=opts).start()
 
-    def get(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None):
+    def get(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None) -> None:
         """Register middleware for HTTP GET requests."""
         return self.method([HttpMethod.GET], *middleware, opts=opts)
 
-    def post(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None):
+    def post(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None) -> None:
         """Register middleware for HTTP POST requests."""
         return self.method([HttpMethod.POST], *middleware, opts=opts)
 
-    def put(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None):
+    def put(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None) -> None:
         """Register middleware for HTTP PUT requests."""
         return self.method([HttpMethod.PUT], *middleware, opts=opts)
 
-    def patch(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None):
+    def patch(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None) -> None:
         """Register middleware for HTTP PATCH requests."""
         return self.method([HttpMethod.PATCH], *middleware, opts=opts)
 
-    def delete(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None):
+    def delete(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None) -> None:
         """Register middleware for HTTP DELETE requests."""
         return self.method([HttpMethod.DELETE], *middleware, opts=opts)
 
-    def options(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None):
+    def options(self, *middleware: HttpMiddleware | HttpHandler, opts: Optional[MethodOptions] = None) -> None:
         """Register middleware for HTTP OPTIONS requests."""
         return self.method([HttpMethod.OPTIONS], *middleware, opts=opts)
 
@@ -361,7 +363,7 @@ def __init__(
         self.server = FunctionServer(ApiWorkerOptions(route.api.name, route.path, methods, opts))
         self.server.http(*route.api.middleware, *route.middleware, *middleware)
 
-    def start(self):
+    def start(self) -> None:
         """Start the server which will respond to incoming requests."""
         Nitric._register_worker(self.server)  # type: ignore
 

nitric/resources/buckets.py

@@ -20,12 +20,12 @@
 
 from nitric.exception import exception_from_grpc_error
 from nitric.api.storage import BucketRef, Storage
-from typing import List, Union
+from typing import List, Union, Callable
 from enum import Enum
 from grpclib import GRPCError
 
 from nitric.application import Nitric
-from nitric.faas import FunctionServer, BucketNotificationWorkerOptions, Middleware, BucketNotificationContext
+from nitric.faas import FunctionServer, BucketNotificationWorkerOptions, BucketNotificationHandler
 from nitric.proto.nitric.resource.v1 import (
     Resource,
     ResourceType,
@@ -87,11 +87,13 @@ def allow(self, *args: Union[BucketPermission, str]) -> BucketRef:
 
         return Storage().bucket(self.name)
 
-    def on(self, notification_type: str, notification_prefix_filter: str):
+    def on(
+        self, notification_type: str, notification_prefix_filter: str
+    ) -> Callable[[BucketNotificationHandler], None]:
         """Create and return a bucket notification decorator for this bucket."""
         print("this has been called")
 
-        def decorator(func: Middleware[BucketNotificationContext]):
+        def decorator(func: BucketNotificationHandler) -> None:
             print("this has been called")
             self._server = FunctionServer(
                 BucketNotificationWorkerOptions(

nitric/resources/topics.py

@@ -20,11 +20,11 @@
 
 from nitric.api.events import Events, TopicRef
 from nitric.exception import exception_from_grpc_error
-from typing import List, Union
+from typing import List, Union, Callable
 from enum import Enum
 from grpclib import GRPCError
 from nitric.application import Nitric
-from nitric.faas import Middleware, FunctionServer, SubscriptionWorkerOptions, EventContext
+from nitric.faas import FunctionServer, SubscriptionWorkerOptions, EventHandler
 from nitric.proto.nitric.resource.v1 import (
     Resource,
     ResourceType,
@@ -80,10 +80,10 @@ def allow(self, *args: Union[TopicPermission, str]) -> TopicRef:
 
         return Events().topic(self.name)
 
-    def subscribe(self):
+    def subscribe(self) -> Callable[[EventHandler], None]:
         """Create and return a subscription decorator for this topic."""
 
-        def decorator(func: Middleware[EventContext]):
+        def decorator(func: EventHandler) -> None:
             server = FunctionServer(SubscriptionWorkerOptions(topic=self.name))
             server.event(func)
             # type ignored because the register call is treated as protected.