Function prime_factorize

python/celery/worker/tasks.py:40–77

(job_id: str, number: int)

@app.task(queue=QUEUE_NAME, name="worker.tasks.prime_factorize")
def prime_factorize(job_id: str, number: int) -> dict[str, Any]:
    original = int(number)
    value = abs(original)
    try:
        if value < 2:
            result = {"number": original, "factors": [original], "is_prime": False}
            update_job(job_id, status="completed", result=result, progress={"step": "Done", "percent": 100})
            return result

        remaining = value
        divisor = 2
        factors: list[int] = []
        max_divisor = max(2, int(value**0.5))

        update_job(job_id, status="running", progress={"step": "Scanning factors", "percent": 0})
        while divisor * divisor <= remaining:
            while remaining % divisor == 0:
                factors.append(divisor)
                remaining //= divisor

            if divisor == 2 or divisor % 100 == 0 or divisor * divisor > remaining:
                update_job(job_id, status="running", progress={
                    "step": f"Testing divisor {divisor}",
                    "percent": min(95, int((divisor / max_divisor) * 100)),
                })
                time.sleep(0.02)

            divisor += 1

        if remaining > 1:
            factors.append(remaining)

        result = {"number": original, "factors": factors, "is_prime": len(factors) == 1}
        update_job(job_id, status="completed", result=result, progress={"step": "Done", "percent": 100})
        return result
    except Exception as exc:
        update_job(job_id, status="failed", result={"error": str(exc)})
        return {"error": str(exc)}