chore: base resource refactor.

Author: tjholm

nitric/resources/apis.py

@@ -124,9 +124,6 @@ def __init__(self, name: str, opts: ApiOptions = None):
         self.security_definitions = opts.security_definitions
         self.security = opts.security
 
-        self._channel = new_default_channel()
-        self._resources_stub = ResourceServiceStub(channel=self._channel)
-
     async def _register(self):
         try:
             await self._resources_stub.declare(

nitric/resources/base.py

@@ -22,7 +22,14 @@
 from abc import ABC, abstractmethod
 from asyncio import Task
 
-from typing import TypeVar, Type, Coroutine, Union
+from typing import TypeVar, Type, Coroutine, Union, List
+
+from grpclib import GRPCError
+from nitricapi.nitric.resource.v1 import Action, PolicyResource, Resource, ResourceType, ResourceDeclareRequest, \
+    ResourceServiceStub
+
+from nitric.api.exception import exception_from_grpc_error
+from nitric.utils import new_default_channel
 
 T = TypeVar("T", bound="BaseResource")
 
@@ -35,6 +42,8 @@ class BaseResource(ABC):
     def __init__(self):
         """Construct a new resource."""
         self._reg: Union[Task, None] = None
+        self._channel = new_default_channel()
+        self._resources_stub = ResourceServiceStub(channel=self._channel)
 
     @abstractmethod
     async def _register(self):
@@ -57,3 +66,41 @@ def make(cls: Type[T], name: str) -> T:
             loop.run_until_complete(r._register())
 
         return r
+
+
+class SecureResource(BaseResource):
+    """A secure base resource class"""
+
+    @abstractmethod
+    def _to_resource(self) -> Resource:
+        pass
+
+    @abstractmethod
+    def _perms_to_actions(self, permissions: List[str]) -> List[Action]:
+        pass
+
+    async def _register_policy_async(self, permissions: List[str]):
+        # if self._reg is not None:
+        #     await asyncio.wait({self._reg})
+
+        policy = PolicyResource(
+            principals=[Resource(type=ResourceType.Function)],
+            actions=self._perms_to_actions(permissions),
+            resources=[self._to_resource()],
+        )
+        try:
+            await self._resources_stub.declare(
+                resource_declare_request=ResourceDeclareRequest(resource=Resource(type=ResourceType.Policy),
+                                                                policy=policy))
+        except GRPCError as grpc_err:
+            raise exception_from_grpc_error(grpc_err)
+        pass
+
+    def _register_policy(self, permissions: List[str]):
+        try:
+            loop = asyncio.get_running_loop()
+            loop.create_task(self._register_policy_async(permissions))
+        except RuntimeError:
+            loop = asyncio.get_event_loop()
+            loop.run_until_complete(self._register_policy_async(permissions))
+

nitric/resources/buckets.py

@@ -36,7 +36,7 @@
     Action, ResourceDeclareRequest,
 )
 
-from nitric.resources.base import BaseResource
+from nitric.resources.base import BaseResource, SecureResource
 
 
 class BucketPermission(Enum):
@@ -47,25 +47,7 @@ class BucketPermission(Enum):
     deleting = "deleting"
 
 
-def _perms_to_actions(permissions: List[Union[BucketPermission, str]]) -> List[Action]:
-    permission_actions_map = {
-        BucketPermission.reading: [Action.BucketFileGet, Action.BucketFileList],
-        BucketPermission.writing: [Action.BucketFilePut],
-        BucketPermission.deleting: [Action.BucketFileDelete],
-    }
-    # convert strings to the enum value where needed
-    perms = [
-        permission if isinstance(permission, BucketPermission) else BucketPermission[permission.lower()]
-        for permission in permissions
-    ]
-    return [action for perm in perms for action in permission_actions_map[perm]]
-
-
-def _to_resource(b: Bucket) -> Resource:
-    return Resource(name=b.name, type=ResourceType.Bucket)
-
-
-class Bucket(BaseResource):
+class Bucket(SecureResource):
     """A bucket resource, used for storage and retrieval of blob/binary data."""
 
     name: str
@@ -75,30 +57,33 @@ def __init__(self, name: str):
         """Create a bucket with the name provided or references it if it already exists."""
         super().__init__()
         self.name = name
-        self._channel = new_default_channel()
-        self._resources_stub = ResourceServiceStub(channel=self._channel)
 
     async def _register(self):
         try:
-            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=_to_resource(self)))
+            await self._resources_stub.declare(
+                resource_declare_request=ResourceDeclareRequest(resource=self._to_resource()))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    async def allow(self, permissions: List[str]) -> BucketRef:
+    def _perms_to_actions(self, permissions: List[Union[BucketPermission, str]]) -> List[Action]:
+        permission_actions_map = {
+            BucketPermission.reading: [Action.BucketFileGet, Action.BucketFileList],
+            BucketPermission.writing: [Action.BucketFilePut],
+            BucketPermission.deleting: [Action.BucketFileDelete],
+        }
+        # convert strings to the enum value where needed
+        perms = [
+            permission if isinstance(permission, BucketPermission) else BucketPermission[permission.lower()]
+            for permission in permissions
+        ]
+        return [action for perm in perms for action in permission_actions_map[perm]]
+
+    def _to_resource(self) -> Resource:
+        return Resource(name=self.name, type=ResourceType.Bucket)
+
+    def allow(self, permissions: List[Union[BucketPermission, str]]) -> BucketRef:
         """Request the required permissions for this resource."""
-        # Ensure registration of the resource is complete before requesting permissions.
-        if self._reg is not None:
-            await self._reg
-
-        policy = PolicyResource(
-            principals=[Resource(type=ResourceType.Function)],
-            actions=_perms_to_actions(permissions),
-            resources=[_to_resource(self)],
-        )
-        try:
-            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=Resource(type=ResourceType.Policy), policy=policy))
-        except GRPCError as grpc_err:
-            raise exception_from_grpc_error(grpc_err)
+        self._register_policy(permissions)
 
         return Storage().bucket(self.name)
 

nitric/resources/collections.py

@@ -36,7 +36,7 @@
     Action, ResourceDeclareRequest,
 )
 
-from nitric.resources.base import BaseResource
+from nitric.resources.base import BaseResource, SecureResource
 
 
 class CollectionPermission(Enum):
@@ -47,56 +47,47 @@ class CollectionPermission(Enum):
     deleting = "deleting"
 
 
-def _perms_to_actions(permissions: List[Union[CollectionPermission, str]]) -> List[Action]:
-    permission_actions_map = {
-        CollectionPermission.reading: [Action.CollectionDocumentRead, Action.CollectionQuery, Action.CollectionList],
-        CollectionPermission.writing: [Action.CollectionDocumentWrite, Action.CollectionList],
-        CollectionPermission.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 permissions
-    ]
 
-    return [action for perm in perms for action in permission_actions_map[perm]]
 
 
-def _to_resource(collection: Collection) -> Resource:
-    return Resource(name=collection.name, type=ResourceType.Collection)
 
 
-class Collection(BaseResource):
+class Collection(SecureResource):
     """A document collection resource."""
 
     def __init__(self, name: str):
         """Construct a new document collection."""
         super().__init__()
         self.name = name
-        self._channel = new_default_channel()
-        self._resources_stub = ResourceServiceStub(channel=self._channel)
 
     async def _register(self):
         try:
-            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=_to_resource(self)))
+            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=self._to_resource()))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    async def allow(self, permissions: List[Union[CollectionPermission, str]]) -> CollectionRef:
+    def _to_resource(self) -> Resource:
+        return Resource(name=self.name, type=ResourceType.Collection)
+
+    def _perms_to_actions(self, permissions: List[Union[CollectionPermission, str]]) -> List[Action]:
+        permission_actions_map = {
+            CollectionPermission.reading: [Action.CollectionDocumentRead, Action.CollectionQuery,
+                                           Action.CollectionList],
+            CollectionPermission.writing: [Action.CollectionDocumentWrite, Action.CollectionList],
+            CollectionPermission.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 permissions
+        ]
+
+        return [action for perm in perms for action in permission_actions_map[perm]]
+
+    def allow(self, permissions: List[Union[CollectionPermission, str]]) -> CollectionRef:
         """Request the required permissions for this collection."""
         # Ensure registration of the resource is complete before requesting permissions.
-        if self._reg is not None:
-            await self._reg
-
-        policy = PolicyResource(
-            principals=[Resource(type=ResourceType.Function)],
-            actions=_perms_to_actions(permissions),
-            resources=[_to_resource(self)],
-        )
-        try:
-            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=Resource(type=ResourceType.Policy), policy=policy))
-        except GRPCError as grpc_err:
-            raise exception_from_grpc_error(grpc_err)
+        self._register_policy(permissions)
 
         return Documents().collection(self.name)
 

nitric/resources/queues.py

@@ -35,7 +35,7 @@
     Action, ResourceDeclareRequest,
 )
 
-from nitric.resources.base import BaseResource
+from nitric.resources.base import BaseResource, SecureResource
 
 
 class QueuePermission(Enum):
@@ -45,25 +45,7 @@ class QueuePermission(Enum):
     receiving = "receiving"
 
 
-def _perms_to_actions(permissions: List[Union[QueuePermission, str]]) -> List[Action]:
-    permission_actions_map = {
-        QueuePermission.sending: [Action.QueueSend, Action.QueueList, Action.QueueDetail],
-        QueuePermission.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 permissions
-    ]
-
-    return [action for perm in perms for action in permission_actions_map[perm]]
-
-
-def _to_resource(queue: Queue) -> Resource:
-    return Resource(name=queue.name, type=ResourceType.Queue)
-
-
-class Queue(BaseResource):
+class Queue(SecureResource):
     """A queue resource."""
 
     name: str
@@ -73,30 +55,34 @@ def __init__(self, name: str):
         """Construct a new queue resource."""
         super().__init__()
         self.name = name
-        self._channel = new_default_channel()
-        self._resources_stub = ResourceServiceStub(channel=self._channel)
+
+    def _to_resource(self) -> Resource:
+        return Resource(name=self.name, type=ResourceType.Queue)
+
+    def _perms_to_actions(self, permissions: List[Union[QueuePermission, str]]) -> List[Action]:
+        permission_actions_map = {
+            QueuePermission.sending: [Action.QueueSend, Action.QueueList, Action.QueueDetail],
+            QueuePermission.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 permissions
+        ]
+
+        return [action for perm in perms for action in permission_actions_map[perm]]
 
     async def _register(self):
         try:
-            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=_to_resource(self)))
+            await self._resources_stub.declare(
+                resource_declare_request=ResourceDeclareRequest(resource=self._to_resource()))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    async def allow(self, permissions: List[Union[QueuePermission, str]]) -> QueueRef:
+    def allow(self, permissions: List[Union[QueuePermission, str]]) -> QueueRef:
         """Request the required permissions for this queue."""
         # Ensure registration of the resource is complete before requesting permissions.
-        if self._reg is not None:
-            await self._reg
-
-        policy = PolicyResource(
-            principals=[Resource(type=ResourceType.Function)],
-            actions=_perms_to_actions(permissions),
-            resources=[_to_resource(self)],
-        )
-        try:
-            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=Resource(type=ResourceType.Policy), policy=policy))
-        except GRPCError as grpc_err:
-            raise exception_from_grpc_error(grpc_err)
+        self._register_policy(permissions)
 
         return Queues().queue(self.name)