Overview

$ python3 -m venv .venv
$ .venv/bin/pip install pytest
[... pip install output ...]
$ .venv/bin/pip show pytest
Name: pytest
Version: 9.1.1
...

"""Capstone: job scheduler -- priority + dependency task ordering.
 
Ingests tasks shaped {id, priority, deps} and emits a valid run order: every
task's dependencies run before it, and among tasks with no remaining
dependencies, the higher-priority task always runs first. Combines a heap
(priority), a dict/set (lookup + seen-tracking), a queue/BFS-style traversal
(dependency relaxation via Kahn's algorithm), and cycle detection in one
runnable program.
 
Big-O per phase (n = number of tasks, e = number of dependency edges):
 
- ``parse_tasks``: O(n) -- one pass building the id -> Task lookup dict.
- ``build_graph``: O(n + e) -- one pass initializing every task's adjacency
  list and in-degree counter, plus one pass per dependency edge to link
  each task to the tasks that depend on it.
- ``schedule`` (Kahn's algorithm, heap-ordered instead of plain-queue-ordered):
  O((n + e) log n) -- every task is pushed and popped from the priority
  heap exactly once (O(log n) per push/pop), and every dependency edge is
  relaxed exactly once, when its source task is emitted.
- Cycle detection: folded into ``schedule`` at O(1) marginal cost -- if the
  emitted order's length is less than the total task count, whatever tasks
  never reached in-degree 0 are stuck in a cycle (or depend on one).
"""
 
from __future__ import annotations
 
import heapq
import json
from dataclasses import dataclass
from pathlib import Path
 
 
class SchedulerCycleError(Exception):
    """Raised when the task graph contains a dependency cycle."""
 
 
@dataclass(frozen=True)
class Task:
    """One schedulable unit of work."""
 
    id: str
    priority: int  # higher number = more urgent; breaks ties among ready tasks
    deps: tuple[str, ...]  # ids of tasks that must complete before this one can run
 
 
def parse_tasks(raw_tasks: list[dict[str, object]]) -> dict[str, Task]:
    """Build an id -> Task lookup from raw dicts -- O(n)."""
    tasks: dict[str, Task] = {}  # => the id -> Task map every later phase queries by id
    for raw in raw_tasks:  # => one pass over the input, O(n)
        task_id = str(raw["id"])
        priority = int(raw["priority"])  # type: ignore[arg-type]
        deps = tuple(str(d) for d in raw.get("deps", []))  # type: ignore[union-attr]
        tasks[task_id] = Task(id=task_id, priority=priority, deps=deps)
    return tasks
 
 
def build_graph(tasks: dict[str, Task]) -> tuple[dict[str, list[str]], dict[str, int]]:
    """Build the adjacency map (dep -> tasks depending on it) and each task's
    in-degree (its unresolved dependency count) -- O(n + e)."""
    adjacency: dict[str, list[str]] = {tid: [] for tid in tasks}  # => O(n) init
    in_degree: dict[str, int] = {tid: 0 for tid in tasks}  # => O(n) init
    for task in tasks.values():  # => O(n) outer pass
        for dep in task.deps:  # => O(e) total across all tasks combined
            if dep not in tasks:  # => O(1) average dict membership check
                raise KeyError(f"{task.id!r} depends on unknown task {dep!r}")
            adjacency[dep].append(task.id)  # => dep unlocks task.id once dep finishes
            in_degree[task.id] += 1  # => task.id has one more unresolved dependency
    return adjacency, in_degree
 
 
def schedule(tasks: dict[str, Task]) -> list[str]:
    """Kahn's topological sort, with a max-priority heap tie-break -- O((n + e) log n).
 
    heapq is a MIN-heap, so pushing (-priority, id) makes the heap always pop
    the HIGHEST-priority ready task first, exactly like Example 41's negation trick.
    """
    adjacency, in_degree = build_graph(tasks)  # => O(n + e)
    ready: list[tuple[int, str]] = [
        (-tasks[tid].priority, tid) for tid in tasks if in_degree[tid] == 0
    ]  # => seeds the heap with every task that has NO dependencies at all
    heapq.heapify(
        ready
    )  # => O(k) where k = len(ready), turns the list into a valid heap
 
    order: list[str] = []  # => the emitted run order, respecting deps and priority ties
    while (
        ready
    ):  # => drains the heap -- each task pushed/popped at most once, O(log n) each
        _, task_id = heapq.heappop(ready)  # => always the highest-priority ready task
        order.append(task_id)
        for dependent in adjacency[
            task_id
        ]:  # => relax every edge OUT of task_id, O(e) total
            in_degree[dependent] -= 1  # => task_id no longer blocks dependent
            if (
                in_degree[dependent] == 0
            ):  # => dependent has NO unresolved deps left -- ready
                heapq.heappush(ready, (-tasks[dependent].priority, dependent))
 
    if len(order) != len(
        tasks
    ):  # => O(1): fewer emissions than tasks means a cycle exists
        stuck = sorted(
            tid for tid in tasks if tid not in order
        )  # => O(n log n), diagnostics only
        raise SchedulerCycleError(
            f"dependency cycle detected among: {', '.join(stuck)}"
        )
    return order  # => a complete, valid run order -- every dep precedes its dependents
 
 
def load_tasks(path: Path) -> dict[str, Task]:
    """Read a JSON file of raw task dicts and parse it into Task objects."""
    with path.open() as f:  # => `with` guarantees the file handle closes
        raw_tasks = json.load(f)  # => parses the whole file as one JSON value
    return parse_tasks(raw_tasks)
 
 
def main() -> None:
    """CLI entry point: schedule the sample tasks.json next to this file."""
    tasks_path = (
        Path(__file__).parent / "tasks.json"
    )  # => resolves relative to THIS file
    tasks = load_tasks(tasks_path)
    order = schedule(tasks)
    print(" -> ".join(order))  # => prints the run order as an arrow-joined chain
 
 
if (
    __name__ == "__main__"
):  # => only runs main() when invoked directly, not when imported
    main()

[
  { "id": "compile", "priority": 3, "deps": [] },
  { "id": "lint", "priority": 5, "deps": [] },
  { "id": "unit_test", "priority": 4, "deps": ["compile"] },
  { "id": "integration_test", "priority": 2, "deps": ["compile", "lint"] },
  { "id": "package", "priority": 1, "deps": ["unit_test", "integration_test"] },
  { "id": "deploy", "priority": 1, "deps": ["package"] }
]

"""pytest coverage for scheduler.py -- an acyclic fixture and a cyclic fixture."""
 
import pytest
 
from scheduler import SchedulerCycleError, parse_tasks, schedule
 
 
def test_acyclic_order_respects_dependencies() -> None:
    """Every dependency must be emitted strictly before every task that depends on it."""
    raw_tasks: list[dict[str, object]] = [
        {"id": "compile", "priority": 3, "deps": []},
        {"id": "lint", "priority": 5, "deps": []},
        {"id": "unit_test", "priority": 4, "deps": ["compile"]},
        {"id": "integration_test", "priority": 2, "deps": ["compile", "lint"]},
        {"id": "package", "priority": 1, "deps": ["unit_test", "integration_test"]},
        {"id": "deploy", "priority": 1, "deps": ["package"]},
    ]
    tasks = parse_tasks(raw_tasks)
    order = schedule(tasks)
 
    assert len(order) == len(tasks)  # every task appears exactly once
    assert set(order) == set(tasks)  # no task is missing or duplicated
    positions = {task_id: index for index, task_id in enumerate(order)}
    for task in tasks.values():
        for dep in task.deps:
            assert positions[dep] < positions[task.id]  # dep runs strictly before task
 
 
def test_acyclic_order_breaks_ties_by_priority() -> None:
    """Among tasks with no remaining dependencies, the higher-priority task runs first."""
    raw_tasks: list[dict[str, object]] = [
        {"id": "compile", "priority": 3, "deps": []},
        {"id": "lint", "priority": 5, "deps": []},
    ]
    tasks = parse_tasks(raw_tasks)
    order = schedule(tasks)
 
    # Both tasks are ready at the same time (no deps); priority 5 beats priority 3.
    assert order == ["lint", "compile"]
 
 
def test_cyclic_graph_raises_scheduler_cycle_error() -> None:
    """A dependency cycle must raise a clear, dedicated error, not silently drop tasks."""
    raw_tasks: list[dict[str, object]] = [
        {"id": "a", "priority": 1, "deps": ["c"]},
        {"id": "b", "priority": 1, "deps": ["a"]},
        {"id": "c", "priority": 1, "deps": ["b"]},  # a -> b -> c -> a: a cycle
    ]
    tasks = parse_tasks(raw_tasks)
 
    with pytest.raises(SchedulerCycleError):
        schedule(tasks)
 
 
def test_cyclic_graph_error_names_the_stuck_tasks() -> None:
    """The raised error message names every task caught in (or blocked by) the cycle."""
    raw_tasks: list[dict[str, object]] = [
        {"id": "a", "priority": 1, "deps": ["b"]},
        {"id": "b", "priority": 1, "deps": ["a"]},
        {"id": "independent", "priority": 1, "deps": []},  # not part of the cycle
    ]
    tasks = parse_tasks(raw_tasks)
 
    with pytest.raises(SchedulerCycleError) as excinfo:
        schedule(tasks)
    assert "a" in str(excinfo.value)
    assert "b" in str(excinfo.value)
    assert "independent" not in str(excinfo.value)  # the healthy task is not implicated

$ .venv/bin/pytest -q
....                                                                     [100%]
4 passed in 0.00s

from scheduler import parse_tasks, schedule, SchedulerCycleError
 
raw = [
    {"id": "a", "priority": 1, "deps": ["b"]},
    {"id": "b", "priority": 1, "deps": ["a"]},
    {"id": "independent", "priority": 1, "deps": []},
]
tasks = parse_tasks(raw)
try:
    schedule(tasks)
except SchedulerCycleError as err:
    print(f"SchedulerCycleError: {err}")

SchedulerCycleError: dependency cycle detected among: a, b

lint -> compile -> unit_test -> integration_test -> package -> deploy