Spaces:
Paused
Paused
File size: 3,811 Bytes
ec17af3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | from __future__ import annotations
import asyncio
import threading
import uuid
from typing import Any
from app.config import settings
_pool = None
_local_jobs: dict[str, dict[str, Any]] = {}
_jobs_lock = threading.Lock()
def _set_local_job(job_id: str, status: str, result: Any = None, error: str | None = None) -> None:
with _jobs_lock:
_local_jobs[job_id] = {
"status": status,
"result": result,
"error": error,
}
# Bound memory to most recent 500 local jobs
if len(_local_jobs) > 500:
oldest_key = next(iter(_local_jobs))
_local_jobs.pop(oldest_key, None)
def _run_coro_in_thread(job_id: str, coro_fn, *args: Any) -> None:
def runner() -> None:
_set_local_job(job_id, "in_progress")
try:
res = asyncio.run(coro_fn({}, *args))
if isinstance(res, dict) and res.get("status") == "failed":
_set_local_job(job_id, "failed", result=None, error=str(res.get("error") or "Job failed"))
else:
_set_local_job(job_id, "complete", result=res, error=None)
except Exception as exc: # pragma: no cover - defensive thread wrapper
_set_local_job(job_id, "failed", result=None, error=str(exc))
thread = threading.Thread(target=runner, daemon=True)
thread.start()
async def get_arq_pool():
global _pool
if _pool is None:
from arq.connections import RedisSettings, create_pool
_pool = await create_pool(
RedisSettings(host=settings.redis_host, port=settings.redis_port, conn_timeout=1, conn_retries=0)
)
return _pool
async def enqueue_transcription(
session_id: int,
user_id: int,
model_name: str | None,
backend_id: str | None,
) -> str:
try:
pool = await get_arq_pool()
job = await pool.enqueue_job(
"transcribe_recording",
session_id,
user_id,
model_name,
backend_id,
)
if job is not None:
return job.job_id
except Exception:
if not settings.use_local_worker_fallback:
raise
from app.worker.jobs import transcribe_recording
job_id = f"local-transcribe-{uuid.uuid4().hex[:12]}"
_set_local_job(job_id, "pending")
_run_coro_in_thread(job_id, transcribe_recording, session_id, user_id, model_name, backend_id)
return job_id
async def enqueue_coaching(session_id: int, user_id: int) -> str:
try:
pool = await get_arq_pool()
job = await pool.enqueue_job("generate_session_coaching", session_id, user_id)
if job is not None:
return job.job_id
except Exception:
if not settings.use_local_worker_fallback:
raise
from app.worker.jobs import generate_session_coaching
job_id = f"local-coaching-{uuid.uuid4().hex[:12]}"
_set_local_job(job_id, "pending")
_run_coro_in_thread(job_id, generate_session_coaching, session_id, user_id)
return job_id
async def get_job_status(job_id: str) -> dict:
with _jobs_lock:
if job_id in _local_jobs:
return dict(_local_jobs[job_id])
try:
pool = await get_arq_pool()
from arq.jobs import Job
job = Job(job_id, pool)
info = await job.info()
if info is None:
return {"status": "pending", "result": None, "error": None}
status_value = info.status.value if info.status else "unknown"
return {
"status": status_value,
"result": info.result if info.success else None,
"error": str(info.result) if not info.success and info.result else None,
}
except Exception:
return {"status": "not_found", "result": None, "error": None}
|