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:

  1. pip install dapr-ext-fastapi

Note

  1. The development package will contain features and behavior that will be compatible with the pre-release version of the Dapr runtime. Make sure to uninstall any stable versions of the Python SDK extension before installing the <code>dapr-dev</code> package.
  1. pip install dapr-ext-fastapi-dev

Example

Subscribing to events of different types

  1. import uvicorn
  2. from fastapi import Body, FastAPI
  3. from dapr.ext.fastapi import DaprApp
  4. from pydantic import BaseModel
  5. class RawEventModel(BaseModel):
  6. body: str
  7. class User(BaseModel):
  8. id: int
  9. name = 'Jane Doe'
  10. class CloudEventModel(BaseModel):
  11. data: User
  12. datacontenttype: str
  13. id: str
  14. pubsubname: str
  15. source: str
  16. specversion: str
  17. topic: str
  18. traceid: str
  19. traceparent: str
  20. tracestate: str
  21. type: str
  22. app = FastAPI()
  23. dapr_app = DaprApp(app)
  24. # Allow handling event with any structure (Easiest, but least robust)
  25. # dapr publish --publish-app-id sample --topic any_topic --pubsub pubsub --data '{"id":"7", "desc": "good", "size":"small"}'
  26. @dapr_app.subscribe(pubsub='pubsub', topic='any_topic')
  27. def any_event_handler(event_data = Body()):
  28. print(event_data)
  29. # For robustness choose one of the below based on if publisher is using CloudEvents
  30. # Handle events sent with CloudEvents
  31. # dapr publish --publish-app-id sample --topic cloud_topic --pubsub pubsub --data '{"id":"7", "name":"Bob Jones"}'
  32. @dapr_app.subscribe(pubsub='pubsub', topic='cloud_topic')
  33. def cloud_event_handler(event_data: CloudEventModel):
  34. print(event_data)
  35. # Handle raw events sent without CloudEvents
  36. # curl -X "POST" http://localhost:3500/v1.0/publish/pubsub/raw_topic?metadata.rawPayload=true -H "Content-Type: application/json" -d '{"body": "345"}'
  37. @dapr_app.subscribe(pubsub='pubsub', topic='raw_topic')
  38. def raw_event_handler(event_data: RawEventModel):
  39. print(event_data)
  40. if __name__ == "__main__":
  41. uvicorn.run(app, host="0.0.0.0", port=30212)

Creating an actor

  1. from fastapi import FastAPI
  2. from dapr.ext.fastapi import DaprActor
  3. from demo_actor import DemoActor
  4. app = FastAPI(title=f'{DemoActor.__name__}Service')
  5. # Add Dapr Actor Extension
  6. actor = DaprActor(app)
  7. @app.on_event("startup")
  8. async def startup_event():
  9. # Register DemoActor
  10. await actor.register_actor(DemoActor)
  11. @app.get("/GetMyData")
  12. def get_my_data():
  13. return "{'message': 'myData'}"