Function send_bulk_emails

examples/celery_alternative/app.py:146–178

Send bulk emails with streaming progress. Celery equivalent: result = send_bulk.apply_async([recipients, subject, body]) while not result.ready(): print(result.info) # Progress polling With dirty arbiters, progress is streamed in real-time!

(data: BulkEmailRequest)

@app.post("/api/email/send-bulk")
async def send_bulk_emails(data: BulkEmailRequest):
    """
    Send bulk emails with streaming progress.

    Celery equivalent:
        result = send_bulk.apply_async([recipients, subject, body])
        while not result.ready():
            print(result.info)  # Progress polling

    With dirty arbiters, progress is streamed in real-time!
    """
    async def generate():
        try:
            client = await get_dirty_client_async()
            async for progress in client.stream_async(
                EMAIL_WORKER,
                "send_bulk_emails",
                recipients=data.recipients,
                subject=data.subject,
                body=data.body,
            ):
                yield f"data: {json.dumps(progress)}\n\n"
        except DirtyError as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )


@app.get("/api/email/stats")