Function aggregate_data

examples/celery_alternative/app.py:277–296

Aggregate data. Celery equivalent: aggregate_data.delay(data, group_by, agg_field, agg_func)

(data: AggregateRequest)

@app.post("/api/data/aggregate")
async def aggregate_data(data: AggregateRequest):
    """
    Aggregate data.

    Celery equivalent:
        aggregate_data.delay(data, group_by, agg_field, agg_func)
    """
    try:
        client = await get_dirty_client_async()
        result = await client.execute_async(
            DATA_WORKER,
            "aggregate",
            data=data.data,
            group_by=data.group_by,
            agg_field=data.agg_field,
            agg_func=data.agg_func,
        )
        return result
    except DirtyError as e:
        raise HTTPException(status_code=500, detail=str(e))


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