(job_id: str, n: int)
| 18 | |
| 19 | @app.task(queue=QUEUE_NAME, name="worker.tasks.calculate_fibonacci") |
| 20 | def calculate_fibonacci(job_id: str, n: int) -> dict[str, Any]: |
| 21 | target = max(0, int(n)) |
| 22 | try: |
| 23 | checkpoints = max(4, min(target or 1, 10)) |
| 24 | for i in range(checkpoints): |
| 25 | update_job(job_id, status="running", progress={ |
| 26 | "step": f"Calculating chunk {i + 1} of {checkpoints}", |
| 27 | "percent": int((i / checkpoints) * 100), |
| 28 | }) |
| 29 | time.sleep(0.25) |
| 30 | |
| 31 | result = {"fibonacci": _fibonacci_number(target), "input": target} |
| 32 | update_job(job_id, status="completed", result=result, progress={"step": "Done", "percent": 100}) |
| 33 | return result |
| 34 | except Exception as exc: |
| 35 | update_job(job_id, status="failed", result={"error": str(exc)}) |
| 36 | return {"error": str(exc)} |
| 37 | |
| 38 | |
| 39 | @app.task(queue=QUEUE_NAME, name="worker.tasks.prime_factorize") |
nothing calls this directly
no test coverage detected