Merge pull request #68 from tjholm/api-refactor

Refactor APIs to match new generation style in betterproto.

Author: tjholm

nitric/api/documents.py

@@ -32,7 +32,8 @@
     Key as KeyMessage,
     Expression as ExpressionMessage,
     ExpressionValue,
-    Document as DocumentMessage,
+    Document as DocumentMessage, DocumentSetRequest, DocumentGetRequest, DocumentDeleteRequest,
+    DocumentQueryStreamRequest, DocumentQueryRequest,
 )
 
 from nitric.utils import new_default_channel, _dict_from_struct, _struct_from_dict
@@ -74,7 +75,9 @@ def collection(self, name: str) -> CollectionRef:
     async def get(self) -> Document:
         """Retrieve the contents of this document, if it exists."""
         try:
-            response = await self._documents._stub.get(key=_doc_ref_to_wire(self))
+            response = await self._documents._stub.get(document_get_request=DocumentGetRequest(
+                key=_doc_ref_to_wire(self)
+            ))
             return _document_from_wire(documents=self._documents, message=response.document)
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
@@ -87,18 +90,20 @@ async def set(self, content: dict):
         """
         try:
             await self._documents._stub.set(
-                key=_doc_ref_to_wire(self),
-                content=_struct_from_dict(content),
+                document_set_request=DocumentSetRequest(
+                    key=_doc_ref_to_wire(self),
+                    content=_struct_from_dict(content),
+                )
             )
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
     async def delete(self):
         """Delete this document, if it exists."""
         try:
-            await self._documents._stub.delete(
+            await self._documents._stub.delete(document_delete_request=DocumentDeleteRequest(
                 key=_doc_ref_to_wire(self),
-            )
+            ))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
@@ -497,9 +502,11 @@ async def stream(self) -> AsyncIterator[Document]:
 
         try:
             async for result in self._documents._stub.query_stream(
-                collection=_collection_to_wire(self._collection),
-                expressions=self._expressions_to_wire(),
-                limit=self._limit,
+                document_query_stream_request=DocumentQueryStreamRequest(
+                    collection=_collection_to_wire(self._collection),
+                    expressions=self._expressions_to_wire(),
+                    limit=self._limit,
+                )
             ):
                 yield _document_from_wire(documents=self._documents, message=result.document)
         except GRPCError as grpc_err:
@@ -513,10 +520,12 @@ async def fetch(self) -> QueryResultsPage:
         """
         try:
             results = await self._documents._stub.query(
-                collection=_collection_to_wire(self._collection),
-                expressions=self._expressions_to_wire(),
-                limit=self._limit,
-                paging_token=self._paging_token,
+                document_query_request=DocumentQueryRequest(
+                    collection=_collection_to_wire(self._collection),
+                    expressions=self._expressions_to_wire(),
+                    limit=self._limit,
+                    paging_token=self._paging_token,
+                )
             )
 
             return QueryResultsPage(

nitric/api/events.py

@@ -24,7 +24,8 @@
 
 from nitric.api.exception import exception_from_grpc_error
 from nitric.utils import new_default_channel, _struct_from_dict
-from nitricapi.nitric.event.v1 import EventServiceStub, NitricEvent, TopicServiceStub
+from nitricapi.nitric.event.v1 import EventServiceStub, NitricEvent, TopicServiceStub, EventPublishRequest, \
+    TopicListRequest
 from dataclasses import dataclass, field
 
 
@@ -70,7 +71,11 @@ async def publish(
             event = Event(**event)
 
         try:
-            response = await self._events._stub.publish(topic=self.name, event=_event_to_wire(event))
+            response = await self._events._stub.publish(
+                event_publish_request=EventPublishRequest(
+                    topic=self.name, event=_event_to_wire(event)
+                )
+            )
             return Event(**{**event.__dict__.copy(), **{"id": response.id}})
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
@@ -97,7 +102,7 @@ def __del__(self):
     async def topics(self) -> List[TopicRef]:
         """Get a list of topics available for publishing or subscription."""
         try:
-            response = await self._topic_stub.list()
+            response = await self._topic_stub.list(topic_list_request=TopicListRequest())
             return [self.topic(topic.name) for topic in response.topics]
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)

nitric/api/queues.py

@@ -24,7 +24,8 @@
 
 from nitric.api.exception import FailedPreconditionException, exception_from_grpc_error, InvalidArgumentException
 from nitric.utils import new_default_channel, _struct_from_dict, _dict_from_struct
-from nitricapi.nitric.queue.v1 import QueueServiceStub, NitricTask, FailedTask as WireFailedTask
+from nitricapi.nitric.queue.v1 import QueueServiceStub, NitricTask, FailedTask as WireFailedTask, QueueCompleteRequest, \
+    QueueSendRequest, QueueSendBatchRequest, QueueReceiveRequest
 from dataclasses import dataclass, field
 
 
@@ -59,7 +60,7 @@ async def complete(self):
                 "Task is missing internal client or lease id, was it returned from " "queue.receive?"
             )
         try:
-            await self._queueing._queue_stub.complete(queue=self._queue.name, lease_id=self.lease_id)
+            await self._queueing._queue_stub.complete(queue_complete_request=QueueCompleteRequest(queue=self._queue.name, lease_id=self.lease_id))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
@@ -149,7 +150,7 @@ async def send(
             task = Task(**task)
 
         try:
-            await self._queueing._queue_stub.send(queue=self.name, task=_task_to_wire(task))
+            await self._queueing._queue_stub.send(queue_send_request=QueueSendRequest(queue=self.name, task=_task_to_wire(task)))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
@@ -167,7 +168,7 @@ async def _send_batch(self, tasks: List[Union[Task, dict]], raise_on_failure: bo
         wire_tasks = [_task_to_wire(Task(**task) if isinstance(task, dict) else task) for task in tasks]
 
         try:
-            response = await self._queueing._queue_stub.send_batch(queue=self.name, tasks=wire_tasks)
+            response = await self._queueing._queue_stub.send_batch(queue_send_batch_request=QueueSendBatchRequest(queue=self.name, tasks=wire_tasks))
             return [_wire_to_failed_task(failed_task) for failed_task in response.failed_tasks]
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
@@ -190,7 +191,7 @@ async def receive(self, limit: int = None) -> List[Task]:
             limit = 1
 
         try:
-            response = await self._queueing._queue_stub.receive(queue=self.name, depth=limit)
+            response = await self._queueing._queue_stub.receive(queue_receive_request=QueueReceiveRequest(queue=self.name, depth=limit))
             # Map the response protobuf response items to Python SDK Nitric Tasks
             return [_wire_to_received_task(task=task, queueing=self._queueing, queue=self) for task in response.tasks]
         except GRPCError as grpc_err:

nitric/api/secrets.py

@@ -24,7 +24,8 @@
 
 from nitric.api.exception import exception_from_grpc_error
 from nitric.utils import new_default_channel
-from nitricapi.nitric.secret.v1 import SecretServiceStub, Secret as SecretMessage, SecretVersion as VersionMessage
+from nitricapi.nitric.secret.v1 import SecretServiceStub, Secret as SecretMessage, SecretVersion as VersionMessage, \
+    SecretPutRequest, SecretAccessRequest
 
 
 class Secrets(object):
@@ -72,7 +73,7 @@ async def put(self, value: Union[str, bytes]) -> SecretVersion:
         secret_message = _secret_to_wire(self)
 
         try:
-            response = await self._secrets._secrets_stub.put(secret=secret_message, value=value)
+            response = await self._secrets._secrets_stub.put(secret_put_request=SecretPutRequest(secret=secret_message, value=value))
             return self.version(version=response.secret_version.version)
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
@@ -111,7 +112,7 @@ async def access(self) -> SecretValue:
         """Return the value stored against this version of the secret."""
         version_message = _secret_version_to_wire(self)
         try:
-            response = await self._secrets._secrets_stub.access(secret_version=version_message)
+            response = await self._secrets._secrets_stub.access(secret_access_request=SecretAccessRequest(secret_version=version_message))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 

nitric/api/storage.py

@@ -22,7 +22,8 @@
 
 from nitric.api.exception import exception_from_grpc_error, InvalidArgumentException
 from nitric.utils import new_default_channel
-from nitricapi.nitric.storage.v1 import StorageServiceStub, StoragePreSignUrlRequestOperation
+from nitricapi.nitric.storage.v1 import StorageServiceStub, StoragePreSignUrlRequestOperation, StorageWriteRequest, \
+    StorageReadRequest, StorageDeleteRequest, StoragePreSignUrlRequest
 from enum import Enum
 
 
@@ -75,7 +76,6 @@ def to_request_operation(self) -> StoragePreSignUrlRequestOperation:
         else:
             raise InvalidArgumentException("Invalid FileMode")
 
-
 @dataclass(frozen=True, order=True)
 class File(object):
     """A reference to a file in a bucket, used to perform operations on that file."""
@@ -91,30 +91,32 @@ async def write(self, body: bytes):
         Will create the file if it doesn't already exist.
         """
         try:
-            await self._storage._storage_stub.write(bucket_name=self._bucket, key=self.key, body=body)
+            await self._storage._storage_stub.write(storage_write_request=StorageWriteRequest(bucket_name=self._bucket, key=self.key, body=body))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
     async def read(self) -> bytes:
         """Read this files contents from the bucket."""
         try:
-            response = await self._storage._storage_stub.read(bucket_name=self._bucket, key=self.key)
+            response = await self._storage._storage_stub.read(storage_read_request=StorageReadRequest(bucket_name=self._bucket, key=self.key))
             return response.body
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
     async def delete(self):
         """Delete this file from the bucket."""
         try:
-            await self._storage._storage_stub.delete(bucket_name=self._bucket, key=self.key)
+            await self._storage._storage_stub.delete(storage_delete_request=StorageDeleteRequest(bucket_name=self._bucket, key=self.key))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
     async def sign_url(self, mode: FileMode = FileMode.READ, expiry: int = 3600):
         """Generate a signed URL for reading or writing to a file."""
         try:
             await self._storage._storage_stub.pre_sign_url(
-                bucket_name=self._bucket, key=self.key, operation=mode.to_request_operation(), expiry=expiry
+                storage_pre_sign_url_request=StoragePreSignUrlRequest(
+                    bucket_name=self._bucket, key=self.key, operation=mode.to_request_operation(), expiry=expiry
+                )
             )
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)

nitric/resources/apis.py

@@ -28,13 +28,11 @@
     ApiResource,
     ApiScopes,
     ApiSecurityDefinition,
-    ApiSecurityDefinitionJwt
+    ApiSecurityDefinitionJwt, ResourceDeclareRequest
 )
 from nitric.utils import new_default_channel
 from grpclib import GRPCError
 from nitric.api.exception import exception_from_grpc_error
-import importlib
-
 
 class JwtSecurityDefinition:
     """Represents the JWT security definition for an API."""
@@ -85,7 +83,7 @@ def _to_resource(b: Api) -> Resource:
     return Resource(name=b.name, type=ResourceType.Api)
 
 
-def security_definition_to_grpc_declaration(security_definitions: SecurityDefinition) -> ApiSecurityDefinition:
+def security_definition_to_grpc_declaration(security_definitions: dict[str,SecurityDefinition]) -> Union[dict[str, ApiSecurityDefinition], None]:
     if security_definitions is None or len(security_definitions) == 0:
         return None
     return {
@@ -132,10 +130,12 @@ def __init__(self, name: str, opts: ApiOptions = None):
     async def _register(self):
         try:
             await self._resources_stub.declare(
-                resource=_to_resource(self), 
-                api=ApiResource(
-                    security_definitions=security_definition_to_grpc_declaration(self.security_definitions), 
-                    security=security_to_grpc_declaration(self.security)
+                resource_declare_request=ResourceDeclareRequest(
+                    resource=_to_resource(self),
+                    api=ApiResource(
+                        security_definitions=security_definition_to_grpc_declaration(self.security_definitions),
+                        security=security_to_grpc_declaration(self.security)
+                    )
                 )
             )
         except GRPCError as grpc_err:

nitric/resources/base.py

@@ -20,6 +20,7 @@
 
 import asyncio
 from abc import ABC, abstractmethod
+from asyncio import Task
 
 from typing import TypeVar, Type, Coroutine, Union
 
@@ -33,7 +34,7 @@ class BaseResource(ABC):
 
     def __init__(self):
         """Construct a new resource."""
-        self._reg: Union[Coroutine, None] = None
+        self._reg: Union[Task, None] = None
 
     @abstractmethod
     async def _register(self):
@@ -48,10 +49,9 @@ def make(cls: Type[T], name: str) -> T:
         """
         # Todo: store the resource reference in a cache to avoid duplicate registrations
         r = cls(name)
-        r._reg = r._register()
         try:
             loop = asyncio.get_running_loop()
-            loop.create_task(r._reg)
+            r._reg = loop.create_task(r._register())
         except RuntimeError:
             loop = asyncio.get_event_loop()
             loop.run_until_complete(r._reg)

nitric/resources/buckets.py

@@ -33,7 +33,7 @@
     ResourceServiceStub,
     PolicyResource,
     ResourceType,
-    Action,
+    Action, ResourceDeclareRequest,
 )
 
 from nitric.resources.base import BaseResource
@@ -80,23 +80,23 @@ def __init__(self, name: str):
 
     async def _register(self):
         try:
-            await self._resources_stub.declare(resource=_to_resource(self))
+            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=_to_resource(self)))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
     async def allow(self, permissions: List[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 asyncio.wait({self._reg})
+            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(policy=policy)
+            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)
 

nitric/resources/collections.py

@@ -33,7 +33,7 @@
     ResourceServiceStub,
     PolicyResource,
     ResourceType,
-    Action,
+    Action, ResourceDeclareRequest,
 )
 
 from nitric.resources.base import BaseResource
@@ -71,29 +71,30 @@ class Collection(BaseResource):
 
     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=_to_resource(self))
+            await self._resources_stub.declare(resource_declare_request=ResourceDeclareRequest(resource=_to_resource(self)))
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
     async 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 asyncio.wait({self._reg})
+            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(policy=policy)
+            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)