dapr-code-generator

DAPR Code Generator

This skill generates production-ready Python code for DAPR applications following best practices.

When to Use

Claude automatically uses this skill when:

• User creates a new DAPR service
• Adding DAPR features to existing code
• Scaffolding microservice projects
• Creating workflows or actors

Code Templates

FastAPI Microservice

"""
{service_name} - DAPR-enabled FastAPI microservice
"""
import json
import logging
from contextlib import asynccontextmanager
from typing import Any

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from dapr.clients import DaprClient
from dapr.ext.fastapi import DaprApp

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Constants
DAPR_STORE_NAME = "statestore"
DAPR_PUBSUB_NAME = "pubsub"


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan handler."""
    logger.info("Starting {service_name}")
    yield
    logger.info("Shutting down {service_name}")


app = FastAPI(
    title="{service_name}",
    lifespan=lifespan
)
dapr_app = DaprApp(app)


# Health endpoints
@app.get("/health")
async def health():
    """Health check endpoint."""
    return {"status": "healthy"}


@app.get("/ready")
async def ready():
    """Readiness check endpoint."""
    return {"status": "ready"}


# DAPR State Management
async def save_state(key: str, value: Any) -> None:
    """Save state to DAPR state store."""
    async with DaprClient() as client:
        await client.save_state(
            store_name=DAPR_STORE_NAME,
            key=key,
            value=json.dumps(value)
        )
        logger.info(f"State saved: {key}")


async def get_state(key: str) -> Any:
    """Get state from DAPR state store."""
    async with DaprClient() as client:
        state = await client.get_state(
            store_name=DAPR_STORE_NAME,
            key=key
        )
        if state.data:
            return json.loads(state.data)
        return None


# DAPR Pub/Sub
async def publish_event(topic: str, data: Any) -> None:
    """Publish event to DAPR pub/sub."""
    async with DaprClient() as client:
        await client.publish_event(
            pubsub_name=DAPR_PUBSUB_NAME,
            topic_name=topic,
            data=json.dumps(data),
            data_content_type="application/json"
        )
        logger.info(f"Event published to {topic}")


# DAPR Service Invocation
async def invoke_service(app_id: str, method: str, data: Any = None) -> Any:
    """Invoke another DAPR service."""
    async with DaprClient() as client:
        response = await client.invoke_method(
            app_id=app_id,
            method_name=method,
            data=json.dumps(data) if data else None,
            content_type="application/json"
        )
        return response.json() if response.data else None


# Example API endpoints - customize as needed
class ItemModel(BaseModel):
    id: str
    name: str
    data: dict = {}


@app.post("/items")
async def create_item(item: ItemModel):
    """Create a new item."""
    await save_state(f"item-{item.id}", item.model_dump())
    await publish_event("items", {"action": "created", "item": item.model_dump()})
    return {"status": "created", "id": item.id}


@app.get("/items/{item_id}")
async def get_item(item_id: str):
    """Get an item by ID."""
    item = await get_state(f"item-{item_id}")
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item


# DAPR Pub/Sub subscription
@dapr_app.subscribe(pubsub=DAPR_PUBSUB_NAME, topic="items")
async def handle_item_event(event: dict):
    """Handle item events from pub/sub."""
    logger.info(f"Received event: {event}")
    return {"status": "SUCCESS"}


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