From 3d0bb563b98c707dd84392489a4ee83016719fcf Mon Sep 17 00:00:00 2001 From: Tim Holm Date: Fri, 23 Jun 2023 16:11:44 +1000 Subject: [PATCH 1/3] Use literals over enums --- nitric/resources/buckets.py | 32 +++++++++++--------------------- nitric/resources/collections.py | 32 ++++++++++---------------------- nitric/resources/queues.py | 28 ++++++++-------------------- nitric/resources/resource.py | 2 +- nitric/resources/secrets.py | 26 ++++++++------------------ nitric/resources/topics.py | 24 ++++++++---------------- 6 files changed, 46 insertions(+), 98 deletions(-) diff --git a/nitric/resources/buckets.py b/nitric/resources/buckets.py index 6937281..5efdfee 100644 --- a/nitric/resources/buckets.py +++ b/nitric/resources/buckets.py @@ -20,7 +20,7 @@ from nitric.exception import exception_from_grpc_error from nitric.api.storage import BucketRef, Storage -from typing import List, Union, Callable +from typing import List, Union, Callable, Literal from enum import Enum from grpclib import GRPCError @@ -35,13 +35,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 +58,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) diff --git a/nitric/resources/collections.py b/nitric/resources/collections.py index f2ac6ac..582bb52 100644 --- a/nitric/resources/collections.py +++ b/nitric/resources/collections.py @@ -20,10 +20,9 @@ from nitric.api.documents import CollectionRef, Documents from nitric.exception import exception_from_grpc_error -from typing import List, Union +from typing import List, Union, Literal from enum import Enum from grpclib import GRPCError - from nitric.application import Nitric from nitric.proto.nitric.resource.v1 import ( Resource, @@ -31,16 +30,10 @@ 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 +53,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] diff --git a/nitric/resources/queues.py b/nitric/resources/queues.py index 5c4c6ad..6ff5889 100644 --- a/nitric/resources/queues.py +++ b/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: diff --git a/nitric/resources/resource.py b/nitric/resources/resource.py index 8ff3a6d..280f755 100644 --- a/nitric/resources/resource.py +++ b/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): diff --git a/nitric/resources/secrets.py b/nitric/resources/secrets.py index f6cc8ce..9cee8f1 100644 --- a/nitric/resources/secrets.py +++ b/nitric/resources/secrets.py @@ -19,7 +19,7 @@ from __future__ import annotations from nitric.exception import exception_from_grpc_error -from typing import List, Union +from typing import List, Union, Literal from enum import Enum from grpclib import GRPCError @@ -34,12 +34,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 +49,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,18 +59,13 @@ 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: """Request the specified permissions to this resource.""" diff --git a/nitric/resources/topics.py b/nitric/resources/topics.py index 4344de8..e7fe7c6 100644 --- a/nitric/resources/topics.py +++ b/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,17 +56,14 @@ 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: """Request the specified permissions to this resource.""" From b886fdfc958d6477f4ad07a2ea21ccd028429e41 Mon Sep 17 00:00:00 2001 From: Tim Holm Date: Fri, 23 Jun 2023 16:15:01 +1000 Subject: [PATCH 2/3] remove remaining str unions on allow --- nitric/resources/queues.py | 2 +- nitric/resources/secrets.py | 2 +- nitric/resources/topics.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/nitric/resources/queues.py b/nitric/resources/queues.py index 6ff5889..b22993d 100644 --- a/nitric/resources/queues.py +++ b/nitric/resources/queues.py @@ -64,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] diff --git a/nitric/resources/secrets.py b/nitric/resources/secrets.py index 9cee8f1..15d4945 100644 --- a/nitric/resources/secrets.py +++ b/nitric/resources/secrets.py @@ -67,7 +67,7 @@ def _perms_to_actions(self, *args: SecretPermission) -> List[int]: 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) diff --git a/nitric/resources/topics.py b/nitric/resources/topics.py index e7fe7c6..9c4cab0 100644 --- a/nitric/resources/topics.py +++ b/nitric/resources/topics.py @@ -65,7 +65,7 @@ def _perms_to_actions(self, *args: TopicPermission) -> List[int]: 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) From 0e622209095faed18e3e8b223fdb5d2277087b9c Mon Sep 17 00:00:00 2001 From: Tim Holm Date: Fri, 23 Jun 2023 16:18:05 +1000 Subject: [PATCH 3/3] remove unused imports --- nitric/resources/buckets.py | 3 +-- nitric/resources/collections.py | 3 +-- nitric/resources/secrets.py | 3 +-- 3 files changed, 3 insertions(+), 6 deletions(-) diff --git a/nitric/resources/buckets.py b/nitric/resources/buckets.py index 5efdfee..5b07832 100644 --- a/nitric/resources/buckets.py +++ b/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, Literal -from enum import Enum +from typing import List, Callable, Literal from grpclib import GRPCError from nitric.application import Nitric diff --git a/nitric/resources/collections.py b/nitric/resources/collections.py index 582bb52..bbc04b0 100644 --- a/nitric/resources/collections.py +++ b/nitric/resources/collections.py @@ -20,8 +20,7 @@ from nitric.api.documents import CollectionRef, Documents from nitric.exception import exception_from_grpc_error -from typing import List, Union, Literal -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 ( diff --git a/nitric/resources/secrets.py b/nitric/resources/secrets.py index 15d4945..b6028cf 100644 --- a/nitric/resources/secrets.py +++ b/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, Literal -from enum import Enum +from typing import List, Literal from grpclib import GRPCError from nitric.application import Nitric