This is the multi-page printable view of this section. Click here to print.

Return to the regular view of this page.

Dapr Python SDK extensions

Python SDK for developing Dapr applications

1 - Getting started with the Dapr Python gRPC service extension

How to get up and running with the Dapr Python gRPC extension

The Dapr Python SDK provides a built in gRPC server extension, dapr.ext.grpc, for creating Dapr services.

Installation

You can download and install the Dapr gRPC server extension with:

pip install "dapr[grpc]"

pip install "dapr[grpc] @ git+https://github.com/dapr/python-sdk.git@main"

Starting with Dapr 1.19, this extension is an extra of the dapr package. The dapr-ext-grpc package is no longer published. If you installed it before, follow the upgrade steps to move to the extra. The import path does not change.

Examples

The App object can be used to create a server.

Listen for service invocation requests

The InvokeMethodReqest and InvokeMethodResponse objects can be used to handle incoming requests.

A simple service that will listen and respond to requests will look like:

from dapr.ext.grpc import App, InvokeMethodRequest, InvokeMethodResponse

app = App()

@app.method(name='my-method')
def mymethod(request: InvokeMethodRequest) -> InvokeMethodResponse:
    print(request.metadata, flush=True)
    print(request.text(), flush=True)

    return InvokeMethodResponse(b'INVOKE_RECEIVED', "text/plain; charset=UTF-8")

app.run(50051)

A full sample can be found here.

Subscribe to a topic

When subscribing to a topic, you can instruct dapr whether the event delivered has been accepted, or whether it should be dropped, or retried later.

Annotate the event parameter of each handler with SubscriptionMessage, the same type that streaming subscriptions deliver. It gives you the parsed payload (data()), the delivery metadata (metadata()) and the CloudEvent fields, such as id(), type() and extensions().

from typing import Optional
from dapr.ext.grpc import App, Rule, SubscriptionMessage
from dapr.clients.grpc._response import TopicEventResponse

app = App()

# Default subscription for a topic
@app.subscribe(pubsub_name='pubsub', topic='TOPIC_A')
def mytopic(event: SubscriptionMessage) -> Optional[TopicEventResponse]:
    print(event.data(),flush=True)
    # Returning None (or not doing a return explicitly) is equivalent
    # to returning a TopicEventResponse("success").
    # You can also return TopicEventResponse("retry") for dapr to log
    # the message and retry delivery later, or TopicEventResponse("drop")
    # for it to drop the message
    return TopicEventResponse("success")

# Specific handler using Pub/Sub routing
@app.subscribe(pubsub_name='pubsub', topic='TOPIC_A',
               rule=Rule("event.type == \"important\"", 1))
def mytopic_important(event: SubscriptionMessage) -> None:
    print(event.data(),flush=True)

# Handler with disabled topic validation
@app.subscribe(pubsub_name='pubsub-mqtt', topic='topic/#', disable_topic_validation=True,)
def mytopic_wildcard(event: SubscriptionMessage) -> None:
    print(event.data(),flush=True)

app.run(50051)

A full sample can be found here.

Setup input binding trigger

from dapr.ext.grpc import App, BindingRequest

app = App()

@app.binding('kafkaBinding')
def binding(request: BindingRequest):
    print(request.text(), flush=True)

app.run(50051)

A full sample can be found here.

2 - Dapr Python SDK integration with FastAPI

How to create Dapr Python virtual actors and pubsub with the FastAPI extension

The Dapr Python SDK provides integration with FastAPI using the dapr.ext.fastapi extension.

Installation

You can download and install the Dapr FastAPI extension with:

pip install "dapr[fastapi]"

pip install "dapr[fastapi] @ git+https://github.com/dapr/python-sdk.git@main"

Starting with Dapr 1.19, this extension is an extra of the dapr package. The dapr-ext-fastapi package is no longer published. If you installed it before, follow the upgrade steps to move to the extra. The import path does not change.

Example

Subscribing to events of different types

import uvicorn
from fastapi import Body, FastAPI
from dapr.ext.fastapi import DaprApp
from pydantic import BaseModel

class RawEventModel(BaseModel):
    body: str

class User(BaseModel):
    id: int
    name: str

class CloudEventModel(BaseModel):
    data: User
    datacontenttype: str
    id: str
    pubsubname: str
    source: str
    specversion: str
    topic: str
    traceid: str
    traceparent: str
    tracestate: str
    type: str    
    
    
app = FastAPI()
dapr_app = DaprApp(app)

# Allow handling event with any structure (Easiest, but least robust)
# dapr publish --publish-app-id sample --topic any_topic --pubsub pubsub --data '{"id":"7", "desc": "good", "size":"small"}'
@dapr_app.subscribe(pubsub='pubsub', topic='any_topic')
def any_event_handler(event_data = Body()):
    print(event_data)    

# For robustness choose one of the below based on if publisher is using CloudEvents

# Handle events sent with CloudEvents
# dapr publish --publish-app-id sample --topic cloud_topic --pubsub pubsub --data '{"id":"7", "name":"Bob Jones"}'
@dapr_app.subscribe(pubsub='pubsub', topic='cloud_topic')
def cloud_event_handler(event_data: CloudEventModel):
    print(event_data)   

# Handle raw events sent without CloudEvents
# curl -X "POST" http://localhost:3500/v1.0/publish/pubsub/raw_topic?metadata.rawPayload=true -H "Content-Type: application/json" -d '{"body": "345"}'
@dapr_app.subscribe(pubsub='pubsub', topic='raw_topic')
def raw_event_handler(event_data: RawEventModel):
    print(event_data)    

 

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=30212)

Creating an actor

from fastapi import FastAPI
from dapr.ext.fastapi import DaprActor
from demo_actor import DemoActor

app = FastAPI(title=f'{DemoActor.__name__}Service')

# Add Dapr Actor Extension
actor = DaprActor(app)

@app.on_event("startup")
async def startup_event():
    # Register DemoActor
    await actor.register_actor(DemoActor)

@app.get("/GetMyData")
def get_my_data():
    return "{'message': 'myData'}"

3 - Dapr Python SDK integration with Flask

How to create Dapr Python virtual actors with the Flask extension

The Dapr Python SDK provides integration with Flask using the dapr.ext.flask extension.

Installation

You can download and install the Dapr Flask extension with:

pip install "dapr[flask]"

pip install "dapr[flask] @ git+https://github.com/dapr/python-sdk.git@main"

Starting with Dapr 1.19, this extension is an extra of the dapr package. The flask-dapr package is no longer published. If you installed it before, follow the upgrade steps to move to the extra.

Example

from flask import Flask
from dapr.ext.flask import DaprActor

from dapr.conf import settings
from demo_actor import DemoActor

app = Flask(f'{DemoActor.__name__}Service')

# Enable DaprActor Flask extension
actor = DaprActor(app)

# Register DemoActor
actor.register_actor(DemoActor)

# Setup method route
@app.route('/GetMyData', methods=['GET'])
def get_my_data():
    return {'message': 'myData'}, 200

# Run application
if __name__ == '__main__':
    app.run(port=settings.HTTP_APP_PORT)

4 - Dapr Python SDK integration with Dapr Workflow extension

How to get up and running with the Dapr Workflow extension

The Dapr Python SDK provides a built-in Dapr Workflow extension, dapr.ext.workflow, for creating Dapr services.

Installation

You can download and install the Dapr Workflow extension with:

pip install "dapr[workflow]"

pip install "dapr[workflow] @ git+https://github.com/dapr/python-sdk.git@main"

Starting with Dapr 1.19, this extension is an extra of the dapr package. The dapr-ext-workflow package is no longer published. If you installed it before, follow the upgrade steps to move to the extra. The import path does not change.

Example

from time import sleep

import dapr.ext.workflow as wf


wfr = wf.WorkflowRuntime()


@wfr.workflow(name='random_workflow')
def task_chain_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
    try:
        result1 = yield ctx.call_activity(step1, input=wf_input)
        result2 = yield ctx.call_activity(step2, input=result1)
    except Exception as e:
        yield ctx.call_activity(error_handler, input=str(e))
        raise
    return [result1, result2]


@wfr.activity(name='step1')
def step1(ctx, activity_input):
    print(f'Step 1: Received input: {activity_input}.')
    # Do some work
    return activity_input + 1


@wfr.activity
def step2(ctx, activity_input):
    print(f'Step 2: Received input: {activity_input}.')
    # Do some work
    return activity_input * 2

@wfr.activity
def error_handler(ctx, error):
    print(f'Executing error handler: {error}.')
    # Do some compensating work


if __name__ == '__main__':
    wfr.start()
    sleep(10)  # wait for workflow runtime to start

    wf_client = wf.DaprWorkflowClient()
    instance_id = wf_client.schedule_new_workflow(workflow=task_chain_workflow, input=42)
    print(f'Workflow started. Instance ID: {instance_id}')
    state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
    print(f'Workflow completed! Status: {state.runtime_status}')

    wfr.shutdown()

Stateful workflow history

From Dapr 1.19, the workflow worker keeps the history of each workflow instance in memory, so the Dapr sidecar sends only the new events on each turn. This is on by default. To turn it off, for example when worker memory is limited, set disable_stateful_history=True on WorkflowRuntime:

import dapr.ext.workflow as wf

wfr = wf.WorkflowRuntime(disable_stateful_history=True)

Learn more in Stateful workflow client.

Waiting for workflows

wait_for_workflow_start and wait_for_workflow_completion ignore the DAPR_API_TIMEOUT_SECONDS environment variable. Without the timeout_in_seconds argument, they wait with no time limit. Pass timeout_in_seconds to limit how long the call waits. When the time limit passes, the call raises TimeoutError:

state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)

Next steps

Getting started with the Dapr Workflow Python SDK

4.1 - Getting started with the Dapr Workflow Python SDK

How to get up and running with workflows using the Dapr Python SDK

Let’s create a Dapr workflow and invoke it using the console. With the provided workflow example, you will:

  • Run a Python console application that demonstrates workflow orchestration with activities, child workflows, and external events
  • Learn how to handle retries, timeouts, and workflow state management
  • Use the Python workflow SDK to start, pause, resume, and purge workflow instances

This example uses the default configuration from dapr init in self-hosted mode.

In the Python example project, the simple.py file contains the setup of the app, including:

  • The workflow definition
  • The workflow activity definitions
  • The registration of the workflow and workflow activities

Prerequisites

Set up the environment

Start by cloning the [Python SDK repo].

git clone https://github.com/dapr/python-sdk.git

From the Python SDK root directory, navigate to the Dapr Workflow example.

cd examples/workflow

Run the following command to install the requirements for running this workflow sample with the Dapr Python SDK.

pip3 install -r workflow/requirements.txt

Run the application locally

To run the Dapr application, you need to start the Python program and a Dapr sidecar. In the terminal, run:

dapr run --app-id wf-simple-example --dapr-grpc-port 50001 --resources-path components -- python3 simple.py

Note: Since Python3.exe is not defined in Windows, you may need to use python simple.py instead of python3 simple.py.

Expected output

- "== APP == Hi Counter!"
- "== APP == New counter value is: 1!"
- "== APP == New counter value is: 11!"
- "== APP == Retry count value is: 0!"
- "== APP == Retry count value is: 1! This print statement verifies retry"
- "== APP == Appending 1 to child_orchestrator_string!"
- "== APP == Appending a to child_orchestrator_string!"
- "== APP == Appending a to child_orchestrator_string!"
- "== APP == Appending 2 to child_orchestrator_string!"
- "== APP == Appending b to child_orchestrator_string!"
- "== APP == Appending b to child_orchestrator_string!"
- "== APP == Appending 3 to child_orchestrator_string!"
- "== APP == Appending c to child_orchestrator_string!"
- "== APP == Appending c to child_orchestrator_string!"
- "== APP == Get response from hello_world_wf after pause call: Suspended"
- "== APP == Get response from hello_world_wf after resume call: Running"
- "== APP == New counter value is: 111!"
- "== APP == New counter value is: 1111!"
- "== APP == Workflow completed! Result: "Completed"

What happened?

When you run the application, several key workflow features are shown:

  1. Workflow and Activity Registration: The application uses Python decorators to automatically register workflows and activities with the runtime. This decorator-based approach provides a clean, declarative way to define your workflow components:

    @wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
    # Workflow definition...

@wfr.activity(name='hello_act')
def hello_act(ctx: WorkflowActivityContext, wf_input):
    # Activity definition...

  2. Runtime Setup: The application initializes the workflow runtime and client:

    wfr = WorkflowRuntime()
wfr.start()
wf_client = DaprWorkflowClient()

  3. Activity Execution: The workflow executes a series of activities that increment a counter:

    @wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
    yield ctx.call_activity(hello_act, input=1)
    yield ctx.call_activity(hello_act, input=10)

  4. Retry Logic: The workflow demonstrates error handling with a retry policy:

    retry_policy = RetryPolicy(
    first_retry_interval=timedelta(seconds=1),
    max_number_of_attempts=3,
    backoff_coefficient=2,
    max_retry_interval=timedelta(seconds=10),
    retry_timeout=timedelta(seconds=100),
)
yield ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)

  5. Child Workflow: A child workflow is executed with its own retry policy:

    yield ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)

  6. External Event Handling: The workflow waits for an external event with a timeout:

    event = ctx.wait_for_external_event(event_name)
timeout = ctx.create_timer(timedelta(seconds=30))
winner = yield when_any([event, timeout])

  7. Workflow Lifecycle Management: The example demonstrates how to pause and resume the workflow:

    wf_client.pause_workflow(instance_id=instance_id)
metadata = wf_client.get_workflow_state(instance_id=instance_id)
# ... check status ...
wf_client.resume_workflow(instance_id=instance_id)

  8. Event Raising: After resuming, the workflow raises an event:

    wf_client.raise_workflow_event(
    instance_id=instance_id,
    event_name=event_name,
    data=event_data
)

  9. Completion and Cleanup: Finally, the workflow waits for completion and cleans up:

    state = wf_client.wait_for_workflow_completion(
    instance_id,
    timeout_in_seconds=30
)
wf_client.purge_workflow(instance_id=instance_id)

List, inspect and rerun workflow instances

From Dapr 1.19, DaprWorkflowClient and the async client in dapr.ext.workflow.aio can list the workflow instances of the app, read the history of an instance, and rerun a finished instance from one of its history events. Use them to find instances without storing their IDs yourself, to see where a failed instance stopped, and to retry it from that step with a corrected input instead of from the start.

import dapr.ext.workflow as wf

wf_client = wf.DaprWorkflowClient()

# Page through the instance IDs of this app, 100 IDs per request
for instance_id in wf_client.iter_workflow_instance_ids(page_size=100):
    print(instance_id)

# Read the history of an instance, oldest event first
history = wf_client.get_workflow_history('order-123')
for event in history:
    print(event.event_id, event.event_type.name, event.name)

# Rerun from the scheduling of the charge_order activity, with a new input
event_id = next(e.event_id for e in history if e.is_rerunnable and e.name == 'charge_order')
new_instance_id = wf_client.rerun_workflow_from_event('order-123', event_id, input=25)
  • list_workflow_instance_ids(page_size=..., continuation_token=...) returns one page with instance_ids and continuation_token. iter_workflow_instance_ids() fetches the pages for you. Listing needs an actor state store that can list keys. Otherwise the call raises NotImplementedError.
  • rerun_workflow_from_event starts a new instance and leaves the source instance unchanged. The source instance must be in a terminal state and must not be a child workflow. The event must be the scheduling of an activity, the creation of a timer or the creation of a child workflow, which is_rerunnable reports. Omit input to keep the recorded input, and pass new_instance_id to choose the ID of the new instance.

For a runnable version, see the workflow management example.

Large workflow payloads

By default, the workflow gRPC channel receives messages of up to 4 MiB, the gRPC default, so a larger message, for example a large workflow input or output, fails with RESOURCE_EXHAUSTED. To raise the limit for sending and receiving, pass max_grpc_message_length in bytes to WorkflowRuntime, DaprWorkflowClient and the async DaprWorkflowClient, or set the DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES environment variable. The argument takes precedence over the environment variable.

wfr = wf.WorkflowRuntime(max_grpc_message_length=16 * 1024 * 1024)
wf_client = wf.DaprWorkflowClient(max_grpc_message_length=16 * 1024 * 1024)

Worker readiness

wfr.start() waits until the worker has opened its work-item stream to the Dapr sidecar, for 30 seconds by default (worker_ready_timeout on WorkflowRuntime). After that, wfr.wait_for_worker_ready(timeout=30.0) returns True when the stream is open, and False if the stream does not open again within the timeout, for example while the worker reconnects after a sidecar restart. Use it in a readiness check, so that the app reports not ready while its worker cannot receive work. Before Python SDK 1.19, it kept returning True after the first connection.

Next steps

5 - Dapr Python SDK integration with Databricks Lakeflow

How to turn Databricks Lakeflow streaming records into durable Dapr Workflow executions

The Dapr Python SDK provides a Databricks Lakeflow integration, dapr.ext.databricks, that turns records emitted by a Lakeflow streaming pipeline into durable Dapr Workflow executions.

External side effects — freezing a card, calling a partner API, opening a case for a human to review — have fundamentally different reliability requirements than transformations inside a data pipeline. Instead of writing that business logic directly inside Lakeflow’s foreach_batch_sink, hand the record to a Dapr Workflow and let it continue independently, with retries, timers, human-in-the-loop waits, and crash recovery, all outside the streaming query’s lifetime.

Databricks Lakeflow
       │
       │ streaming records
       ▼
Dapr Databricks Sink
       │
       ▼
Dapr Workflow
       │
  ┌────┼──────────────┐
  ▼    ▼              ▼
 APIs  SaaS          Humans
       systems

Installation

You can download and install the Dapr Databricks extension with:

pip install "dapr[workflow,databricks]"

Example

from pyspark import pipelines as dp
from dapr.ext.databricks import register_workflow_sink

register_workflow_sink(
    name='order_actions',
    workflow='process_order',
    id_field='order_id',
    namespace='orders',
)

@dp.append_flow(target='order_actions', name='order_actions_flow')
def order_actions_flow():
    return spark.readStream.table('validated_orders')

register_workflow_sink registers the foreach_batch_sink for you — you do not write one by hand for the standard case. Every workflow instance ID is derived deterministically from namespace/name/generation and the record’s own business key (id_field, never a random UUID), so a Lakeflow retry of the same micro-batch never starts a second workflow execution for the same record.

Next steps

Getting started with the Dapr Databricks Python SDK extension

Learn more:

5.1 - Getting started with the Dapr Databricks Python SDK extension

How to turn Databricks Lakeflow streaming records into durable Dapr Workflow executions

Let’s run a fraud-remediation example that turns simulated Databricks Lakeflow records into durable Dapr Workflow executions. With the provided example, you will:

  • Run a Dapr Workflow (fraud_remediation) that freezes a card, notifies the customer, opens an investigation, waits for an analyst decision, and resolves the case
  • Feed it two simulated Lakeflow micro-batches through dapr.ext.databricks’s batch handler — no Databricks workspace or pyspark required for this part
  • See a retried micro-batch schedule zero duplicate workflow executions, because every instance ID is derived deterministically from the record’s own business key
Databricks Lakeflow identifies a suspicious transaction
                |
         dapr.ext.databricks sink
                |
        FraudRemediationWorkflow
                |
  freeze card -> notify customer -> open investigation
                |
         wait for analyst decision
                |
            resolve case

This example uses the default configuration from dapr init in self-hosted mode.

Prerequisites

Set up the environment

Clone the Python SDK repo:

git clone https://github.com/dapr/python-sdk.git

From the repo root, go to the Databricks example:

cd examples/databricks

This example has two files:

  • fraud_remediation_workflow.py — the Dapr Workflow side: the fraud_remediation workflow and its activities. Its __main__ block simulates a small Lakeflow micro-batch in-process using DaprWorkflowBatchHandler directly — the same building block register_workflow_sink uses internally — then retries the identical batch to demonstrate that no duplicate workflow executions are created. This is the file you’ll run below.
  • fraud_remediation_pipeline.py — the Databricks Lakeflow side: register_workflow_sink plus the @dp.append_flow that feeds it. This file is illustrative only — it depends on pyspark.pipelines (provided by the Databricks Lakeflow runtime, not pip install pyspark) and a Unity Catalog table, so it can’t run locally. Copy its pattern into an actual Databricks Lakeflow pipeline to wire the integration up for real.

Run the application locally

To run the example, start the Python program and a Dapr sidecar together:

dapr run --app-id fraud-remediation-demo -- python3 fraud_remediation_workflow.py

Expected output (interleaved with Dapr/durabletask log lines):

*** Lakeflow micro-batch 1: scheduling fraud remediation workflows
*** Micro-batch 1 handed off; each workflow now runs independently
*** Simulating a Lakeflow retry of the same micro-batch
*** Retry complete: no duplicate workflow executions were created
*** Triggered by sink 'fraud_actions', batch 1
*** Freezing card for transaction T-1001 (customer C-1)
*** Notifying customer C-1 about the frozen card
*** Opened investigation CASE-T-1001
*** Resolved CASE-T-1001: CONFIRMED_FRAUD
*** Workflow fraud-fraud_actions-v1-T-1001 completed: {"case_id": "CASE-T-1001", "decision": "CONFIRMED_FRAUD"}
... (and the same for T-1002)

Look for the structured dapr.ext.databricks: sink=... outcome=... log lines: the first micro-batch reports outcome=newly_scheduled for both transactions, and the simulated retry reports outcome=already_existed for both — the same transaction never starts a second remediation workflow.

When you’re done:

dapr stop --app-id fraud-remediation-demo

What happened?

  1. register_workflow_sink (used inside the example’s batch handler, and the function you’d call from a real @dp.append_flow pipeline) computes a deterministic workflow instance ID for every record — <namespace>-<sink>-<generation>-<business_key> — never a random UUID.
  2. For each record, the handler checks whether a workflow instance with that ID already exists. If not, it schedules one; if so, it treats the record as already handled. Both the first micro-batch and the simulated retry go through this same check, which is why the retry logs outcome=already_existed instead of starting a second fraud_remediation execution.
  3. Once scheduled, each workflow instance runs independently of the Lakeflow batch that created it — freeze card -> notify customer -> open investigation -> wait for an analyst decision -> resolve case continues even after the streaming micro-batch has moved on.
  4. If neither the “does it exist” check nor the schedule call can complete (Dapr unavailable, a timeout, an auth failure), the whole micro-batch fails so Lakeflow retries it — the sink never reports success while a record’s fate is unknown.

This is retry-safe, not exactly-once: it depends on Dapr retaining that instance’s history/state. See the delivery semantics and full-refresh section of the extension’s reference docs for what that means for your workflow-history retention, and how generation lets you deliberately replay business actions after a full Lakeflow pipeline refresh.

Next steps

  • Copy fraud_remediation_pipeline.py’s pattern into a real Databricks Lakeflow pipeline to wire this up against real streaming data.
  • See the full configuration reference — composite business keys, custom input mapping, bounded concurrency, and observability.
  • Learn more about authoring the workflow side: Dapr Workflow.