text: # -*- coding: utf-8 -*-
  
  """
  Copyright (c) Microsoft Corporation and Dapr Contributors.
  Licensed under the MIT License.
  """
  
  from typing import Callable, Dict, List, Optional, Union
  
  from dapr.clients.base import DaprActorClientBase
  from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_UNKNOWN
  from dapr.clients.grpc.client import DaprGrpcClient, MetadataTuple, InvokeMethodResponse
  from dapr.clients.http.dapr_actor_http_client import DaprActorHttpClient
  from dapr.clients.http.dapr_invocation_http_client import DaprInvocationHttpClient
  from dapr.conf import settings
  from google.protobuf.message import Message as GrpcMessage
  
  __all__ = [
      'DaprClient',
      'DaprActorClientBase',
      'DaprActorHttpClient',
      'DaprInternalError',
      'ERROR_CODE_UNKNOWN',
  ]
  
  from grpc import (  # type: ignore
      UnaryUnaryClientInterceptor,
      UnaryStreamClientInterceptor,
      StreamUnaryClientInterceptor,
      StreamStreamClientInterceptor
  )
  
  
  class DaprClient(DaprGrpcClient):
      """The Dapr python-sdk uses gRPC for most operations. The exception being
      service invocation which needs to support HTTP to HTTP invocations. The sdk defaults
      to HTTP but can be overridden with the DAPR_API_METHOD_INVOCATION_PROTOCOL environment
      variable. See: https://github.com/dapr/python-sdk/issues/176 for more details"""
  
      def __init__(
          self,
          address: Optional[str] = None,
          headers_callback: Optional[Callable[[], Dict[str, str]]] = None,
          interceptors: Optional[List[Union[
              UnaryUnaryClientInterceptor,
              UnaryStreamClientInterceptor,
              StreamUnaryClientInterceptor,
              StreamStreamClientInterceptor]]] = None):
          """Connects to Dapr Runtime and via gRPC and HTTP.
  
          Args:
              address (str, optional): Dapr Runtime gRPC endpoint address.
              headers_callback (lambda: Dict[str, str]], optional): Generates header for each request.
              interceptors (list of UnaryUnaryClientInterceptor or
                  UnaryStreamClientInterceptor or
                  StreamUnaryClientInterceptor or
                  StreamStreamClientInterceptor, optional): gRPC interceptors.
          """
          super().__init__(address, interceptors)
          self.invocation_client = None
  
          invocation_protocol = settings.DAPR_API_METHOD_INVOCATION_PROTOCOL.upper()
  
          if invocation_protocol == 'HTTP':
              self.invocation_client = DaprInvocationHttpClient(headers_callback=headers_callback)
          elif invocation_protocol == 'GRPC':
              pass
          else:
              raise DaprInternalError(
                  f'Unknown value for DAPR_API_METHOD_INVOCATION_PROTOCOL: {invocation_protocol}')
  
      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:
          """Invoke a service method over gRPC or HTTP.
  
          Args:
              app_id (str): Application Id.
              method_name (str): Method to be invoked.
              data (bytes or str or GrpcMessage, optional): Data for requet's body.
              content_type (str, optional): Content type of the data.
              metadata (MetadataTuple, optional): Additional metadata or headers.
              http_verb (str, optional): HTTP verb for the request.
              http_querystring (MetadataTuple, optional): Query parameters.
  
          Returns:
              InvokeMethodResponse: the response from the method invocation.
          """
          if self.invocation_client:
              return self.invocation_client.invoke_method(
                  app_id,
                  method_name,
                  data,
                  content_type=content_type,
                  metadata=metadata,
                  http_verb=http_verb,
                  http_querystring=http_querystring)
          else:
              return super().invoke_method(
                  app_id,
                  method_name,
                  data,
                  content_type=content_type,
                  metadata=metadata,
                  http_verb=http_verb,
                  http_querystring=http_querystring)
  