Function run_etl

examples/celery_alternative/app.py:300–323

Run ETL pipeline with streaming progress. Celery equivalent: chain(extract.s(), transform.s(), load.s()).apply_async()

(data: ETLRequest)

@app.post("/api/data/etl")
async def run_etl(data: ETLRequest):
    """
    Run ETL pipeline with streaming progress.

    Celery equivalent:
        chain(extract.s(), transform.s(), load.s()).apply_async()
    """
    async def generate():
        try:
            client = await get_dirty_client_async()
            async for progress in client.stream_async(
                DATA_WORKER,
                "etl_pipeline",
                source_data=data.source_data,
                transformations=data.transformations,
            ):
                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",
    )


@app.post("/api/data/query")