Add an HTTP client for service invocation and default to using it

.github/workflows/validate_examples.yaml

@@ -19,11 +19,11 @@ jobs:
       GOOS: linux
       GOARCH: amd64
       GOPROXY: https://proxy.golang.org
-      DAPR_CLI_VER: 1.0.0-rc.3
-      DAPR_RUNTIME_VER: 1.0.0-rc.2
-      DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/3dacfb672d55f1436c249057aaebbe597e1066f3/install/install.sh
+      DAPR_CLI_VER: 1.0.0-rc.4
+      DAPR_RUNTIME_VER: 1.0.0-rc.3
+      DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.0.0-rc.4/install/install.sh
       DAPR_CLI_REF:
-      DAPR_REF: f4327d91c3cd921b26077d99abe0f237c9a38f5d
+      DAPR_REF: 
     steps:
       - uses: actions/checkout@v2
       - name: Set up Python ${{ env.PYTHON_VER }}
@@ -82,4 +82,4 @@ jobs:
           ./dist/linux_amd64/release/placement --healthz-port 9091 &
       - name: Check Examples
         run: |
-          tox -e examples
+          tox -e examples

dapr/clients/__init__.py

@@ -7,7 +7,7 @@
 
 from dapr.clients.base import DaprActorClientBase
 from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_UNKNOWN
-from dapr.clients.grpc.client import DaprClient
+from dapr.clients.main import DaprClient
 from dapr.clients.http.dapr_actor_http_client import DaprActorHttpClient
 
 __all__ = [

dapr/clients/grpc/_response.py

@@ -191,6 +191,11 @@ def unpack(self, message: GrpcMessage) -> None:
             ValueError: message is not protocol buffer message object or message's type is not
                 matched with the response data type
         """
+
+        if self.content_type is not None and self.content_type.lower() == "application/x-protobuf":
+            message.ParseFromString(self.data)
+            return
+
         unpack(self.proto, message)
 
 

dapr/clients/grpc/client.py

@@ -38,7 +38,7 @@
 from opencensus.trace.tracers.base import Tracer   # type: ignore
 
 
-class DaprClient:
+class DaprGrpcClient:
     """The convenient layer implementation of Dapr gRPC APIs.
 
     This provides the wrappers and helpers to allows developers to use Dapr runtime gRPC API
@@ -86,7 +86,7 @@ def close(self):
     def __del__(self):
         self.close()
 
-    def __enter__(self) -> 'DaprClient':
+    def __enter__(self) -> 'DaprGrpcClient':
         return self
 
     def __exit__(self, exc_type, exc_value, traceback) -> None:

dapr/clients/http/client.py

@@ -0,0 +1,86 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT License.
+"""
+
+import aiohttp
+
+from typing import Dict, Optional, Union, Tuple, TYPE_CHECKING
+if TYPE_CHECKING:
+    from dapr.serializers import Serializer
+
+from dapr.conf import settings
+from dapr.clients.base import DEFAULT_JSON_CONTENT_TYPE
+from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_DOES_NOT_EXIST, ERROR_CODE_UNKNOWN
+from opencensus.trace.tracers.base import Tracer   # type: ignore
+
+CONTENT_TYPE_HEADER = 'content-type'
+DAPR_API_TOKEN_HEADER = 'dapr-api-token'
+
+
+class DaprHttpClient:
+    """A Dapr Http API client"""
+
+    def __init__(self,
+                 message_serializer: 'Serializer',
+                 timeout: int = 60,
+                 tracer: Optional[Tracer] = None):
+
+        self._timeout = aiohttp.ClientTimeout(total=timeout)
+        self._serializer = message_serializer
+        self._tracer = tracer
+
+    def get_api_url(self) -> str:
+        return 'http://{}:{}/{}'.format(
+            settings.DAPR_RUNTIME_HOST,
+            settings.DAPR_HTTP_PORT,
+            settings.DAPR_API_VERSION)
+
+    async def send_bytes(
+            self, method: str, url: str,
+            data: Optional[bytes],
+            headers: Dict[str, Union[bytes, str]] = {},
+            query_params: Dict[str, Union[bytes, str]] = {}
+    ) -> Tuple[bytes, aiohttp.ClientResponse]:
+        if not headers.get(CONTENT_TYPE_HEADER):
+            headers[CONTENT_TYPE_HEADER] = DEFAULT_JSON_CONTENT_TYPE
+
+        if settings.DAPR_API_TOKEN is not None:
+            headers[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN
+
+        if self._tracer is not None:
+            trace_headers = self._tracer.propagator.to_headers(self._tracer.span_context)
+            headers.update(trace_headers)
+
+        r = None
+        async with aiohttp.ClientSession(timeout=self._timeout) as session:
+            r = await session.request(
+                method=method,
+                url=url,
+                data=data,
+                headers=headers,
+                params=query_params)
+
+            if r.status >= 200 and r.status < 300:
+                return await r.read(), r
+
+            raise (await self.convert_to_error(r))
+
+    async def convert_to_error(self, response) -> DaprInternalError:
+        error_info = None
+        try:
+            error_body = await response.read()
+            if (error_body is None or len(error_body) == 0) and response.status == 404:
+                return DaprInternalError("Not Found", ERROR_CODE_DOES_NOT_EXIST)
+            error_info = self._serializer.deserialize(error_body)
+        except Exception:
+            return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}')
+
+        if error_info and isinstance(error_info, dict):
+            message = error_info.get('message')
+            error_code = error_info.get('errorCode') or ERROR_CODE_UNKNOWN
+            return DaprInternalError(message, error_code)
+
+        return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}')

dapr/clients/http/dapr_actor_http_client.py

@@ -4,26 +4,27 @@
 Copyright (c) Microsoft Corporation.
 Licensed under the MIT License.
 """
-import aiohttp
 
-from typing import Dict, Optional, TYPE_CHECKING
+from typing import Optional, TYPE_CHECKING
+
 if TYPE_CHECKING:
     from dapr.serializers import Serializer
 
-from dapr.conf import settings
-from dapr.clients.base import DaprActorClientBase, DEFAULT_JSON_CONTENT_TYPE
-from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_DOES_NOT_EXIST, ERROR_CODE_UNKNOWN
-
-
-CONTENT_TYPE_HEADER = 'content-type'
+from dapr.clients.http.client import DaprHttpClient
+from dapr.clients.base import DaprActorClientBase
+from opencensus.trace.tracers.base import Tracer  # type: ignore
 
 
 class DaprActorHttpClient(DaprActorClientBase):
     """A Dapr Actor http client implementing :class:`DaprActorClientBase`"""
 
-    def __init__(self, message_serializer: 'Serializer', timeout: int = 60):
-        self._timeout = aiohttp.ClientTimeout(total=timeout)
-        self._serializer = message_serializer
+    def __init__(
+            self,
+            message_serializer: 'Serializer',
+            timeout: int = 60,
+            tracer: Optional[Tracer] = None):
+
+        self._client = DaprHttpClient(message_serializer, timeout, tracer)
 
     async def invoke_method(
             self, actor_type: str, actor_id: str,
@@ -40,7 +41,9 @@ async def invoke_method(
             bytes: the response from the actor.
         """
         url = f'{self._get_base_url(actor_type, actor_id)}/method/{method}'
-        return await self._send_bytes(method='POST', url=url, data=data)
+
+        body, r = await self._client.send_bytes(method='POST', url=url, data=data)
+        return body
 
     async def save_state_transactionally(
             self, actor_type: str, actor_id: str,
@@ -53,7 +56,7 @@ async def save_state_transactionally(
             data (bytes): Json-serialized the transactional state operations.
         """
         url = f'{self._get_base_url(actor_type, actor_id)}/state'
-        await self._send_bytes(method='PUT', url=url, data=data)
+        await self._client.send_bytes(method='PUT', url=url, data=data)
 
     async def get_state(
             self, actor_type: str, actor_id: str, name: str) -> bytes:
@@ -68,7 +71,8 @@ async def get_state(
             bytes: the value of the state.
         """
         url = f'{self._get_base_url(actor_type, actor_id)}/state/{name}'
-        return await self._send_bytes(method='GET', url=url, data=None)
+        body, r = await self._client.send_bytes(method='GET', url=url, data=None)
+        return body
 
     async def register_reminder(
             self, actor_type: str, actor_id: str, name: str, data: bytes) -> None:
@@ -81,7 +85,7 @@ async def register_reminder(
             data (bytes): Reminder request json body.
         """
         url = f'{self._get_base_url(actor_type, actor_id)}/reminders/{name}'
-        await self._send_bytes(method='PUT', url=url, data=data)
+        await self._client.send_bytes(method='PUT', url=url, data=data)
 
     async def unregister_reminder(
             self, actor_type: str, actor_id: str, name: str) -> None:
@@ -93,7 +97,7 @@ async def unregister_reminder(
             name (str):  the name of reminder.
         """
         url = f'{self._get_base_url(actor_type, actor_id)}/reminders/{name}'
-        await self._send_bytes(method='DELETE', url=url, data=None)
+        await self._client.send_bytes(method='DELETE', url=url, data=None)
 
     async def register_timer(
             self, actor_type: str, actor_id: str, name: str, data: bytes) -> None:
@@ -106,7 +110,7 @@ async def register_timer(
             data (bytes): Timer request json body.
         """
         url = f'{self._get_base_url(actor_type, actor_id)}/timers/{name}'
-        await self._send_bytes(method='PUT', url=url, data=data)
+        await self._client.send_bytes(method='PUT', url=url, data=data)
 
     async def unregister_timer(
             self, actor_type: str, actor_id: str, name: str) -> None:
@@ -118,44 +122,10 @@ async def unregister_timer(
             name (str): The name of timer
         """
         url = f'{self._get_base_url(actor_type, actor_id)}/timers/{name}'
-        await self._send_bytes(method='DELETE', url=url, data=None)
+        await self._client.send_bytes(method='DELETE', url=url, data=None)
 
     def _get_base_url(self, actor_type: str, actor_id: str) -> str:
-        return 'http://{}:{}/{}/actors/{}/{}'.format(
-            settings.DAPR_RUNTIME_HOST,
-            settings.DAPR_HTTP_PORT,
-            settings.DAPR_API_VERSION,
+        return '{}/actors/{}/{}'.format(
+            self._client.get_api_url(),
             actor_type,
             actor_id)
-
-    async def _send_bytes(
-            self, method: str, url: str,
-            data: Optional[bytes], headers: Dict[str, str] = {}) -> bytes:
-        if not headers.get(CONTENT_TYPE_HEADER):
-            headers[CONTENT_TYPE_HEADER] = DEFAULT_JSON_CONTENT_TYPE
-
-        r = None
-        async with aiohttp.ClientSession(timeout=self._timeout) as session:
-            r = await session.request(method=method, url=url, data=data, headers=headers)
-
-        if r.status >= 200 and r.status < 300:
-            return await r.read()
-
-        raise (await self.convert_to_error(r))
-
-    async def convert_to_error(self, response) -> DaprInternalError:
-        error_info = None
-        try:
-            error_body = await response.read()
-            if (error_body is None or len(error_body) == 0) and response.status == 404:
-                return DaprInternalError("Not Found", ERROR_CODE_DOES_NOT_EXIST)
-            error_info = self._serializer.deserialize(error_body)
-        except Exception:
-            return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}')
-
-        if error_info and isinstance(error_info, dict):
-            message = error_info.get('message')
-            error_code = error_info.get('errorCode') or ERROR_CODE_UNKNOWN
-            return DaprInternalError(message, error_code)
-
-        return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}')

dapr/clients/http/dapr_invocation_http_client.py

@@ -0,0 +1,76 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT License.
+"""
+
+import asyncio
+
+from typing import Optional, Union
+
+from dapr.clients.http.client import DaprHttpClient, CONTENT_TYPE_HEADER
+from dapr.clients.grpc._helpers import MetadataTuple, GrpcMessage
+from dapr.clients.grpc._response import InvokeMethodResponse
+from dapr.serializers import DefaultJSONSerializer
+from opencensus.trace.tracers.base import Tracer   # type: ignore
+
+
+class DaprInvocationHttpClient:
+    """Service Invocation HTTP Client"""
+
+    def __init__(self, timeout: int = 60, tracer: Optional[Tracer] = None):
+        self._client = DaprHttpClient(DefaultJSONSerializer(), timeout, tracer)
+
+    def invoke_method(
+            self,
+            app_id: str,
+            method_name: str,
+            data: Union[bytes, str, GrpcMessage],
+            content_type: Optional[str] = None,
+            metadata: Optional[MetadataTuple] = None,
+            http_verb: Optional[str] = None,
+            http_querystring: Optional[MetadataTuple] = None) -> InvokeMethodResponse:
+        """This implementation is designed to seemlessly mimic the behavior of the grpc
+        service invoke implementation. Please see the grpc version for parameter definitions
+        and samples"""
+
+        verb = 'GET'
+        if http_verb is not None:
+            verb = http_verb
+
+        headers = {}
+        if metadata is not None:
+            for key, value in metadata:
+                headers[key] = value
+        query_params = {}
+        if http_querystring is not None:
+            for key, value in http_querystring:
+                query_params[key] = value
+
+        if content_type is not None:
+            headers[CONTENT_TYPE_HEADER] = content_type
+
+        url = f'{self._client.get_api_url()}/invoke/{app_id}/method/{method_name}'
+
+        if isinstance(data, GrpcMessage):
+            body = data.SerializeToString()
+        elif isinstance(data, str):
+            body = data.encode('utf-8')
+        else:
+            body = data
+
+        async def make_request():
+            resp_body, r = await self._client.send_bytes(
+                method=verb,
+                headers=headers,
+                url=url,
+                data=body,
+                query_params=query_params)
+
+            resp_data = InvokeMethodResponse(resp_body, r.content_type)
+            for key in r.headers:
+                resp_data.headers[key] = r.headers.getall(key)  # type: ignore
+            return resp_data
+
+        return asyncio.run(make_request())