22358373981

Committed 24 Feb 2026 03:47PM UTC coverage: 68.113%. First build

Build # 22358373981

Build Type

Pull #1403

github

Committed by

web-flow

Commit Message

Merge 9b6420310 into 7506dd15c

Pull Request Pull Request #1403: Runner Manager support

Run Details

632 of 1453 new or added lines in 24 files covered. (43.5%)

12812 of 18810 relevant lines covered (68.11%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.05

/pod/video_encode_transcript/task_queue.py

"""Queue ranking helpers for encoding/transcription task dispatch in Esup-Pod.

The queue is intentionally simple:
- only pending tasks receive a rank,
- rank is recomputed from scratch to keep ordering deterministic,
- non-pending tasks must not retain stale rank values.
"""

import logging
from typing import TypeAlias

from pod.video.models import Video

from .models import Task

log = logging.getLogger(__name__)

QueuePriority: TypeAlias = int
HIGH_PRIORITY: QueuePriority = 1
LOW_PRIORITY: QueuePriority = 2


def get_user_priority(video: Video) -> QueuePriority:
    """Return queue priority for a video owner."""
    try:
        affiliation = video.owner.owner.affiliation
        if affiliation == "student":
            return LOW_PRIORITY
        return HIGH_PRIORITY
    except AttributeError:
        log.warning(
            "Cannot determine affiliation for video %s owner. Defaulting to high priority.",
            video.id,
        )
        return HIGH_PRIORITY
    except Exception as exc:
        log.warning(
            "Error getting affiliation for video %s: %s. Defaulting to high priority.",
            video.id,
            str(exc),
        )
        return HIGH_PRIORITY


def _task_priority(task: Task) -> QueuePriority:
    """Return priority for a pending task."""
    if not task.video_id or not task.video:
        return HIGH_PRIORITY
    return get_user_priority(task.video)


def get_sorted_pending_tasks() -> list[Task]:
    """Return all pending tasks sorted by queue priority and creation date."""
    pending_tasks = list(
        Task.objects.filter(status="pending")
        .select_related("video", "video__owner", "video__owner__owner")
        .order_by("date_added", "id")
    )
    pending_tasks.sort(
        key=lambda task: (_task_priority(task), task.date_added, task.id),
    )
    return pending_tasks


def refresh_pending_task_ranks() -> None:
    """Recalculate rank for all pending tasks and clear rank for non-pending ones."""
    sorted_tasks = get_sorted_pending_tasks()
    updates: list[Task] = []
    for index, task in enumerate(sorted_tasks, start=1):
        if task.rank != index:
            task.rank = index
            updates.append(task)

    if updates:
        # Batch write only changed rows to avoid unnecessary UPDATE queries.
        Task.objects.bulk_update(updates, ["rank"])

    # Keep DB consistent: only pending tasks are expected to have a queue rank.
    Task.objects.exclude(status="pending").exclude(rank__isnull=True).update(rank=None)


def get_video_pending_encoding_queue_info(video: Video) -> tuple[int | None, int]:
    """Return (rank, total_pending) for the encoding task linked to a video."""
    queue_total = Task.objects.filter(status="pending").count()
    queue_task = (
        Task.objects.filter(video_id=video.id, type="encoding", status="pending")
        .order_by("date_added", "id")
        .first()
    )
    queue_rank = queue_task.rank if queue_task else None
    return queue_rank, queue_total

1	"""Queue ranking helpers for encoding/transcription task dispatch in Esup-Pod.
2
3	The queue is intentionally simple:
4	- only pending tasks receive a rank,
5	- rank is recomputed from scratch to keep ordering deterministic,
6	- non-pending tasks must not retain stale rank values.
7	"""
8
9	import logging	1✔
10	from typing import TypeAlias	1✔
11
12	from pod.video.models import Video	1✔
13
14	from .models import Task	1✔
15
16	log = logging.getLogger(__name__)	1✔
17
18	QueuePriority: TypeAlias = int	1✔
19	HIGH_PRIORITY: QueuePriority = 1	1✔
20	LOW_PRIORITY: QueuePriority = 2	1✔
21
22
23	def get_user_priority(video: Video) -> QueuePriority:	1✔
24	"""Return queue priority for a video owner."""
25	try:	1✔
26	affiliation = video.owner.owner.affiliation	1✔
27	if affiliation == "student":	1✔
28	return LOW_PRIORITY	1✔
29	return HIGH_PRIORITY	1✔
NEW 30	except AttributeError:	×
NEW 31	log.warning(	×
32	"Cannot determine affiliation for video %s owner. Defaulting to high priority.",
33	video.id,
34	)
NEW 35	return HIGH_PRIORITY	×
NEW 36	except Exception as exc:	×
NEW 37	log.warning(	×
38	"Error getting affiliation for video %s: %s. Defaulting to high priority.",
39	video.id,
40	str(exc),
41	)
NEW 42	return HIGH_PRIORITY	×
43
44
45	def _task_priority(task: Task) -> QueuePriority:	1✔
46	"""Return priority for a pending task."""
47	if not task.video_id or not task.video:	1✔
48	return HIGH_PRIORITY	1✔
49	return get_user_priority(task.video)	1✔
50
51
52	def get_sorted_pending_tasks() -> list[Task]:	1✔
53	"""Return all pending tasks sorted by queue priority and creation date."""
54	pending_tasks = list(	1✔
55	Task.objects.filter(status="pending")
56	.select_related("video", "video__owner", "video__owner__owner")
57	.order_by("date_added", "id")
58	)
59	pending_tasks.sort(	1✔
60	key=lambda task: (_task_priority(task), task.date_added, task.id),
61	)
62	return pending_tasks	1✔
63
64
65	def refresh_pending_task_ranks() -> None:	1✔
66	"""Recalculate rank for all pending tasks and clear rank for non-pending ones."""
67	sorted_tasks = get_sorted_pending_tasks()	1✔
68	updates: list[Task] = []	1✔
69	for index, task in enumerate(sorted_tasks, start=1):	1✔
70	if task.rank != index:	1✔
71	task.rank = index	1✔
72	updates.append(task)	1✔
73
74	if updates:	1✔
75	# Batch write only changed rows to avoid unnecessary UPDATE queries.
76	Task.objects.bulk_update(updates, ["rank"])	1✔
77
78	# Keep DB consistent: only pending tasks are expected to have a queue rank.
79	Task.objects.exclude(status="pending").exclude(rank__isnull=True).update(rank=None)	1✔
80
81
82	def get_video_pending_encoding_queue_info(video: Video) -> tuple[int \| None, int]:	1✔
83	"""Return (rank, total_pending) for the encoding task linked to a video."""
84	queue_total = Task.objects.filter(status="pending").count()	1✔
85	queue_task = (	1✔
86	Task.objects.filter(video_id=video.id, type="encoding", status="pending")
87	.order_by("date_added", "id")
88	.first()
89	)
90	queue_rank = queue_task.rank if queue_task else None	1✔
91	return queue_rank, queue_total	1✔

EsupPortail / Esup-Pod / 22358373981

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous