chore: Replace permission enums with literals (#113)

Author: tjholm

nitric/resources/buckets.py

@@ -20,8 +20,7 @@
 
 from nitric.exception import exception_from_grpc_error
 from nitric.api.storage import BucketRef, Storage
-from typing import List, Union, Callable
-from enum import Enum
+from typing import List, Callable, Literal
 from grpclib import GRPCError
 
 from nitric.application import Nitric
@@ -35,13 +34,7 @@
 
 from nitric.resources.resource import SecureResource
 
-
-class BucketPermission(Enum):
-    """Valid query expression operators."""
-
-    reading = "reading"
-    writing = "writing"
-    deleting = "deleting"
+BucketPermission = Literal["reading", "writing", "deleting"]
 
 
 class Bucket(SecureResource):
@@ -64,23 +57,19 @@ async def _register(self):
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    def _perms_to_actions(self, *args: Union[BucketPermission, str]) -> List[Action]:
-        permission_actions_map = {
-            BucketPermission.reading: [Action.BucketFileGet, Action.BucketFileList],
-            BucketPermission.writing: [Action.BucketFilePut],
-            BucketPermission.deleting: [Action.BucketFileDelete],
+    def _perms_to_actions(self, *args: BucketPermission) -> List[int]:
+        permission_actions_map: dict[BucketPermission, List[int]] = {
+            "reading": [Action.BucketFileGet, Action.BucketFileList],
+            "writing": [Action.BucketFilePut],
+            "deleting": [Action.BucketFileDelete],
         }
-        # convert strings to the enum value where needed
-        perms = [
-            permission if isinstance(permission, BucketPermission) else BucketPermission[permission.lower()]
-            for permission in args
-        ]
-        return [action for perm in perms for action in permission_actions_map[perm]]
+
+        return [action for perm in args for action in permission_actions_map[perm]]
 
     def _to_resource(self) -> Resource:
-        return Resource(name=self.name, type=ResourceType.Bucket)
+        return Resource(name=self.name, type=ResourceType.Bucket) # type:ignore
 
-    def allow(self, *args: Union[BucketPermission, str]) -> BucketRef:
+    def allow(self, *args: BucketPermission) -> BucketRef:
         """Request the required permissions for this resource."""
         str_args = [str(permission) for permission in args]
         self._register_policy(*str_args)

nitric/resources/collections.py

@@ -20,27 +20,19 @@
 
 from nitric.api.documents import CollectionRef, Documents
 from nitric.exception import exception_from_grpc_error
-from typing import List, Union
-from enum import Enum
+from typing import List, Literal
 from grpclib import GRPCError
-
 from nitric.application import Nitric
 from nitric.proto.nitric.resource.v1 import (
     Resource,
     ResourceType,
     Action,
     ResourceDeclareRequest,
 )
-
 from nitric.resources.resource import SecureResource
 
 
-class CollectionPermission(Enum):
-    """Valid query expression operators."""
-
-    reading = "reading"
-    writing = "writing"
-    deleting = "deleting"
+CollectionPermission = Literal["reading", "writing", "deleting"]
 
 
 class Collection(SecureResource):
@@ -60,27 +52,22 @@ async def _register(self):
             raise exception_from_grpc_error(grpc_err)
 
     def _to_resource(self) -> Resource:
-        return Resource(name=self.name, type=ResourceType.Collection)
+        return Resource(name=self.name, type=ResourceType.Collection) # type:ignore
 
-    def _perms_to_actions(self, *args: Union[CollectionPermission, str]) -> List[Action]:
-        permission_actions_map = {
-            CollectionPermission.reading: [
+    def _perms_to_actions(self, *args: CollectionPermission) -> List[int]:
+        permission_actions_map: dict[CollectionPermission, List[int]] = {
+            "reading": [
                 Action.CollectionDocumentRead,
                 Action.CollectionQuery,
                 Action.CollectionList,
             ],
-            CollectionPermission.writing: [Action.CollectionDocumentWrite, Action.CollectionList],
-            CollectionPermission.deleting: [Action.CollectionDocumentDelete, Action.CollectionList],
+            "writing": [Action.CollectionDocumentWrite, Action.CollectionList],
+            "deleting": [Action.CollectionDocumentDelete, Action.CollectionList],
         }
-        # convert strings to the enum value where needed
-        perms = [
-            permission if isinstance(permission, CollectionPermission) else CollectionPermission[permission.lower()]
-            for permission in args
-        ]
 
-        return [action for perm in perms for action in permission_actions_map[perm]]
+        return [action for perm in args for action in permission_actions_map[perm]]
 
-    def allow(self, *args: Union[CollectionPermission, str]) -> CollectionRef:
+    def allow(self, *args: CollectionPermission) -> CollectionRef:
         """Request the required permissions for this collection."""
         # Ensure registration of the resource is complete before requesting permissions.
         str_args = [str(permission) for permission in args]

nitric/resources/queues.py

@@ -19,8 +19,7 @@
 from __future__ import annotations
 
 from nitric.exception import exception_from_grpc_error
-from typing import List, Union
-from enum import Enum
+from typing import List, Union, Literal
 from grpclib import GRPCError
 from nitric.api.queues import QueueRef, Queues
 from nitric.application import Nitric
@@ -33,13 +32,7 @@
 
 from nitric.resources.resource import SecureResource
 
-
-class QueuePermission(Enum):
-    """Valid query expression operators."""
-
-    sending = "sending"
-    receiving = "receiving"
-
+QueuePermission = Literal["sending", "receiving"]
 
 class Queue(SecureResource):
     """A queue resource."""
@@ -53,20 +46,15 @@ def __init__(self, name: str):
         self.name = name
 
     def _to_resource(self) -> Resource:
-        return Resource(name=self.name, type=ResourceType.Queue)
+        return Resource(name=self.name, type=ResourceType.Queue) # type:ignore
 
-    def _perms_to_actions(self, *args: Union[QueuePermission, str]) -> List[Action]:
-        permission_actions_map = {
-            QueuePermission.sending: [Action.QueueSend, Action.QueueList, Action.QueueDetail],
-            QueuePermission.receiving: [Action.QueueReceive, Action.QueueList, Action.QueueDetail],
+    def _perms_to_actions(self, *args: QueuePermission) -> List[int]:
+        permission_actions_map: dict[QueuePermission, List[int]] = {
+            "sending": [Action.QueueSend, Action.QueueList, Action.QueueDetail],
+            "receiving": [Action.QueueReceive, Action.QueueList, Action.QueueDetail],
         }
-        # convert strings to the enum value where needed
-        perms = [
-            permission if isinstance(permission, QueuePermission) else QueuePermission[permission.lower()]
-            for permission in args
-        ]
 
-        return [action for perm in perms for action in permission_actions_map[perm]]
+        return [action for perm in args for action in permission_actions_map[perm]]
 
     async def _register(self):
         try:
@@ -76,7 +64,7 @@ async def _register(self):
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    def allow(self, *args: Union[QueuePermission, str]) -> QueueRef:
+    def allow(self, *args: QueuePermission) -> QueueRef:
         """Request the required permissions for this queue."""
         # Ensure registration of the resource is complete before requesting permissions.
         str_args = [str(permission) for permission in args]

nitric/resources/resource.py

@@ -79,7 +79,7 @@ def _to_resource(self) -> WireResource:
         pass
 
     @abstractmethod
-    def _perms_to_actions(self, *args: str) -> List[Action]:
+    def _perms_to_actions(self, *args: Any) -> List[int]:
         pass
 
     async def _register_policy_async(self, *args: str):

nitric/resources/secrets.py

@@ -19,8 +19,7 @@
 from __future__ import annotations
 
 from nitric.exception import exception_from_grpc_error
-from typing import List, Union
-from enum import Enum
+from typing import List, Literal
 from grpclib import GRPCError
 
 from nitric.application import Nitric
@@ -34,12 +33,7 @@
 
 from nitric.resources.resource import SecureResource
 
-
-class SecretPermission(Enum):
-    """Available permissions that can be requested for secret resources."""
-
-    accessing = "accessing"
-    putting = "putting"
+SecretPermission = Literal["accessing", "putting"]
 
 
 class Secret(SecureResource):
@@ -54,7 +48,7 @@ def __init__(self, name: str):
         self.name = name
 
     def _to_resource(self) -> Resource:
-        return Resource(name=self.name, type=ResourceType.Secret)
+        return Resource(name=self.name, type=ResourceType.Secret) # type:ignore
 
     async def _register(self):
         try:
@@ -64,20 +58,15 @@ async def _register(self):
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    def _perms_to_actions(self, *args: Union[SecretPermission, str]) -> List[Action]:
-        permissions_actions_map = {
-            SecretPermission.accessing: [Action.SecretAccess],
-            SecretPermission.putting: [Action.SecretPut],
+    def _perms_to_actions(self, *args: SecretPermission) -> List[int]:
+        permissions_actions_map: dict[SecretPermission, List[int]] = {
+            "accessing": [Action.SecretAccess],
+            "putting": [Action.SecretPut],
         }
-        # convert strings to the enum value where needed
-        perms = [
-            permission if isinstance(permission, SecretPermission) else SecretPermission[permission.lower()]
-            for permission in args
-        ]
 
-        return [action for perm in perms for action in permissions_actions_map[perm]]
+        return [action for perm in args for action in permissions_actions_map[perm]]
 
-    def allow(self, *args: Union[SecretPermission, str]) -> SecretContainerRef:
+    def allow(self, *args: SecretPermission) -> SecretContainerRef:
         """Request the specified permissions to this resource."""
         str_args = [str(permission) for permission in args]
         self._register_policy(*str_args)

nitric/resources/topics.py

@@ -20,8 +20,7 @@
 
 from nitric.api.events import Events, TopicRef
 from nitric.exception import exception_from_grpc_error
-from typing import List, Union, Callable
-from enum import Enum
+from typing import List, Union, Callable, Literal
 from grpclib import GRPCError
 from nitric.application import Nitric
 from nitric.faas import FunctionServer, SubscriptionWorkerOptions, EventHandler
@@ -34,11 +33,7 @@
 
 from nitric.resources.resource import SecureResource
 
-
-class TopicPermission(Enum):
-    """Valid query expression operators."""
-
-    publishing = "publishing"
+TopicPermission = Literal["publishing"]
 
 
 class Topic(SecureResource):
@@ -61,19 +56,16 @@ async def _register(self):
             raise exception_from_grpc_error(grpc_err)
 
     def _to_resource(self) -> Resource:
-        return Resource(name=self.name, type=ResourceType.Topic)
+        return Resource(name=self.name, type=ResourceType.Topic) # type:ignore
 
-    def _perms_to_actions(self, *args: Union[TopicPermission, str]) -> List[Action]:
-        _permMap = {TopicPermission.publishing: [Action.TopicEventPublish]}
-        # convert strings to the enum value where needed
-        perms = [
-            permission if isinstance(permission, TopicPermission) else TopicPermission[permission.lower()]
-            for permission in args
-        ]
+    def _perms_to_actions(self, *args: TopicPermission) -> List[int]:
+        _permMap: dict[TopicPermission, List[int]] = {
+            "publishing": [Action.TopicEventPublish]
+        }
 
-        return [action for perm in perms for action in _permMap[perm]]
+        return [action for perm in args for action in _permMap[perm]]
 
-    def allow(self, *args: Union[TopicPermission, str]) -> TopicRef:
+    def allow(self, *args: TopicPermission) -> TopicRef:
         """Request the specified permissions to this resource."""
         str_args = [str(permission) for permission in args]
         self._register_policy(*str_args)