Purge all tasks in the Celery queue. Only accessible to authenticated users.
(current_user: User = Depends(get_current_user))
| 191 | ## Purge all tasks in the Celery queue |
| 192 | @router.post("/purge-queue", tags=["Queue Management"], description="Purge all tasks in the Celery queue", summary="Purge all tasks in the Celery queue") |
| 193 | async def purge_queue(current_user: User = Depends(get_current_user)): |
| 194 | """ |
| 195 | Purge all tasks in the Celery queue. |
| 196 | Only accessible to authenticated users. |
| 197 | """ |
| 198 | if not current_user: # Add your own authentication checks |
| 199 | raise HTTPException(status_code=403, detail="Not authorized to purge queue") |
| 200 | |
| 201 | try: |
| 202 | purge_celery_queue() |
| 203 | return {"status": "success", "message": "Celery queue purged successfully"} |
| 204 | except Exception as e: |
| 205 | return {"status": "error", "message": str(e)} |
nothing calls this directly
no test coverage detected