Getting started with the Dapr client Python SDK

How to get up and running with the Dapr Python SDK

The Dapr client package allows you to interact with other Dapr applications from a Python application.

Pre-requisites

Import the client package

The dapr package contains the DaprClient which will be used to create and use a client.

  1. from dapr.clients import DaprClient

Building blocks

The Python SDK allows you to interface with all of the Dapr building blocks.

Invoke a service

  1. from dapr.clients import DaprClient
  3. with DaprClient() as d:
  4.     # invoke a method (gRPC or HTTP GET)    
  5.     resp = d.invoke_method('service-to-invoke', 'method-to-invoke', data='{"message":"Hello World"}')
  7.     # for other HTTP verbs the verb must be specified
  8.     # invoke a 'POST' method (HTTP only)    
  9.     resp = d.invoke_method('service-to-invoke', 'method-to-invoke', data='{"id":"100", "FirstName":"Value", "LastName":"Value"}', http_verb='post')

Save & get application state

  1. from dapr.clients import DaprClient
  3. with DaprClient() as d:
  4.     # Save state
  5.     d.save_state(store_name="statestore", key="key1", value="value1")
  7.     # Get state
  8.     data = d.get_state(store_name="statestore", key="key1").data
  10.     # Delete state
  11.     d.delete_state(store_name="statestore", key="key1")

Query application state (Alpha)

  1.     from dapr import DaprClient
  3.     query = '''
  4.     {
  5.         "filter": {
  6.             "EQ": { "state": "CA" }
  7.         },
  8.         "sort": [
  9.             {
  10.                 "key": "person.id",
  11.                 "order": "DESC"
  12.             }
  13.         ]
  14.     }
  15.     '''
  17.     with DaprClient() as d:
  18.         resp = d.query_state(
  19.             store_name='state_store',
  20.             query=query,
  21.             states_metadata={"metakey": "metavalue"},  # optional
  22.         )

Publish & subscribe to messages

Publish messages
  1. from dapr.clients import DaprClient
  3. with DaprClient() as d:
  4.     resp = d.publish_event(pubsub_name='pubsub', topic_name='TOPIC_A', data='{"message":"Hello World"}')
Subscribe to messages
  1. from cloudevents.sdk.event import v1
  2. from dapr.ext.grpc import App
  3. import json
  5. app = App()
  7. # Default subscription for a topic
  8. @app.subscribe(pubsub_name='pubsub', topic='TOPIC_A')
  9. def mytopic(event: v1.Event) -> None:
  10.     data = json.loads(event.Data())
  11.     print(f'Received: id={data["id"]}, message="{data ["message"]}"' 
  12.           ' content_type="{event.content_type}"',flush=True)
  14. # Specific handler using Pub/Sub routing
  15. @app.subscribe(pubsub_name='pubsub', topic='TOPIC_A',
  16.                rule=Rule("event.type == \"important\"", 1))
  17. def mytopic_important(event: v1.Event) -> None:
  18.     data = json.loads(event.Data())
  19.     print(f'Received: id={data["id"]}, message="{data ["message"]}"' 
  20.           ' content_type="{event.content_type}"',flush=True)

Interact with output bindings

  1. from dapr.clients import DaprClient
  3. with DaprClient() as d:
  4.     resp = d.invoke_binding(binding_name='kafkaBinding', operation='create', data='{"message":"Hello World"}')

Retrieve secrets

  1. from dapr.clients import DaprClient
  3. with DaprClient() as d:
  4.     resp = d.get_secret(store_name='localsecretstore', key='secretKey')

Get configuration

  1. from dapr.clients import DaprClient
  3. with DaprClient() as d:
  4.     # Get Configuration
  5.     configuration = d.get_configuration(store_name='configurationstore', keys=['orderId'], config_metadata={})

Subscribe to configuration

  1. import asyncio
  2. from time import sleep
  3. from dapr.clients import DaprClient
  5. async def executeConfiguration():
  6.     with DaprClient() as d:
  7.         storeName = 'configurationstore'
  9.         key = 'orderId'
  11.         # Wait for sidecar to be up within 20 seconds.
  12.         d.wait(20)
  14.         # Subscribe to configuration by key.
  15.         configuration = await d.subscribe_configuration(store_name=storeName, keys=[key], config_metadata={})
  16.         while True:
  17.             if configuration != None:
  18.                 items = configuration.get_items()
  19.                 for key, item in items:
  20.                     print(f"Subscribe key={key} value={item.value} version={item.version}", flush=True)
  21.             else:
  22.                 print("Nothing yet")
  23.         sleep(5)
  25. asyncio.run(executeConfiguration())

Distributed Lock

  1. from dapr.clients import DaprClient
  3. def main():
  4.     # Lock parameters
  5.     store_name = 'lockstore'  # as defined in components/lockstore.yaml
  6.     resource_id = 'example-lock-resource'
  7.     client_id = 'example-client-id'
  8.     expiry_in_seconds = 60
  10.     with DaprClient() as dapr:
  11.         print('Will try to acquire a lock from lock store named [%s]' % store_name)
  12.         print('The lock is for a resource named [%s]' % resource_id)
  13.         print('The client identifier is [%s]' % client_id)
  14.         print('The lock will will expire in %s seconds.' % expiry_in_seconds)
  16.         with dapr.try_lock(store_name, resource_id, client_id, expiry_in_seconds) as lock_result:
  17.             assert lock_result.success, 'Failed to acquire the lock. Aborting.'
  18.             print('Lock acquired successfully!!!')
  20.         # At this point the lock was released - by magic of the `with` clause ;)
  21.         unlock_result = dapr.unlock(store_name, resource_id, client_id)
  22.         print('We already released the lock so unlocking will not work.')
  23.         print('We tried to unlock it anyway and got back [%s]' % unlock_result.status)

Workflow

  1. from dapr.ext.workflow import WorkflowRuntime, DaprWorkflowContext, WorkflowActivityContext
  2. from dapr.clients import DaprClient
  4. instanceId = "exampleInstanceID"
  5. workflowComponent = "dapr"
  6. workflowName = "hello_world_wf"
  7. eventName = "event1"
  8. eventData = "eventData"
  10. def main():
  11.     with DaprClient() as d:
  12.         host = settings.DAPR_RUNTIME_HOST
  13.         port = settings.DAPR_GRPC_PORT
  14.         workflowRuntime = WorkflowRuntime(host, port)
  15.         workflowRuntime = WorkflowRuntime()
  16.         workflowRuntime.register_workflow(hello_world_wf)
  17.         workflowRuntime.register_activity(hello_act)
  18.         workflowRuntime.start()
  20.         # Start the workflow
  21.         start_resp = d.start_workflow(instance_id=instanceId, workflow_component=workflowComponent,
  22.                         workflow_name=workflowName, input=inputData, workflow_options=workflowOptions)
  23.         print(f"start_resp {start_resp.instance_id}")
  25.         # ...
  27.         # Pause Test
  28.         d.pause_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  29.         getResponse = d.get_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  30.         print(f"Get response from {workflowName} after pause call: {getResponse.runtime_status}")
  32.         # Resume Test
  33.         d.resume_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  34.         getResponse = d.get_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  35.         print(f"Get response from {workflowName} after resume call: {getResponse.runtime_status}")
  37.         sleep(1)
  38.         # Raise event
  39.         d.raise_workflow_event(instance_id=instanceId, workflow_component=workflowComponent,
  40.                     event_name=eventName, event_data=eventData)
  42.         sleep(5)
  43.         # Purge Test
  44.         d.purge_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  45.         try:
  46.             getResponse = d.get_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  47.         except DaprInternalError as err:
  48.             if nonExistentIDError in err._message:
  49.                 print("Instance Successfully Purged")
  52.         # Kick off another workflow for termination purposes 
  53.         # This will also test using the same instance ID on a new workflow after
  54.         # the old instance was purged
  55.         start_resp = d.start_workflow(instance_id=instanceId, workflow_component=workflowComponent,
  56.                         workflow_name=workflowName, input=inputData, workflow_options=workflowOptions)
  57.         print(f"start_resp {start_resp.instance_id}")
  59.         # Terminate Test
  60.         d.terminate_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  61.         sleep(1)
  62.         getResponse = d.get_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  63.         print(f"Get response from {workflowName} after terminate call: {getResponse.runtime_status}")
  65.         # Purge Test
  66.         d.purge_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  67.         try:
  68.             getResponse = d.get_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  69.         except DaprInternalError as err:
  70.             if nonExistentIDError in err._message:
  71.                 print("Instance Successfully Purged")
  73.         workflowRuntime.shutdown()