Scheduling API

🔥 Vibe Prompt

"Build a FastAPI job scheduling service using Earliest Deadline First strategy."

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Scheduling API")

class Job(BaseModel):
    id: str
    deadline: int
    profit: int
    duration: int

@app.post("/schedule")
def schedule_jobs(req: list[Job]):
    sorted_jobs = sorted(req, key=lambda j: j.deadline)
    scheduled = []
    current_time = 0
    for job in sorted_jobs:
        if current_time + job.duration <= job.deadline:
            scheduled.append(job)
            current_time += job.duration
    return {"scheduled": [j.id for j in scheduled], "total_profit": sum(j.profit for j in scheduled)}

Greedy Course Complete! 🎉

✅ Kruskal/Prim MST
✅ Huffman Coding
✅ Set Cover
✅ Scheduling API

Chapter Summary

Understand core concepts and principles
Master implementation methods and techniques
Familiar with common issues and solutions
Able to apply in real projects

Implementation Example

Basic Example

# This section provides a complete implementation example

Steps

Setup: Configure development environment
Data: Prepare required data
Implementation: Build core functionality
Testing: Verify correctness
Optimization: Improve performance

Common Errors

| Error Type | Cause | Solution | |------------|-------|----------| | Compilation | Syntax | Check code syntax | | Runtime | Environment | Verify dependencies installed | | Logic | Algorithm | Step-by-step debugging | | Performance | Efficiency | Use profilers |

Code Example

import sys

def main():
    print("Hello, World!")

if __name__ == "__main__":
    main()

References

Official documentation
API reference
Open source examples
Community discussions

Greedy Scheduling Algorithms

| Algorithm | Strategy | Objective | |-----------|----------|-----------| | Earliest Deadline First (EDF) | Sort by deadline | Minimize max lateness | | Shortest Processing Time (SPT) | Sort by duration | Minimize average completion | | Shortest Remaining Time (SRT) | Preemptive, shortest remaining | Minimize average waiting | | Round Robin | Fixed time slices | Fair time-sharing | | Priority Scheduling | Highest priority first | Meet priority requirements |

Earliest Deadline First (EDF)

Minimizes maximum lateness (how far past the deadline a job finishes).

def earliest_deadline_first(jobs):
    """
    Schedule jobs by earliest deadline.
    
    Args:
        jobs: list of (name, duration, deadline)
    
    Returns:
        schedule: ordered list of job names
        max_lateness: how far past deadline the latest job finishes
    """
    sorted_jobs = sorted(jobs, key=lambda j: j[2])  # Sort by deadline
    schedule = []
    current_time = 0
    max_lateness = 0
    
    for name, duration, deadline in sorted_jobs:
        schedule.append(name)
        current_time += duration
        lateness = max(0, current_time - deadline)
        max_lateness = max(max_lateness, lateness)
    
    return schedule, max_lateness


# Example
jobs = [
    ("Job A", 3, 5),
    ("Job B", 2, 3),
    ("Job C", 4, 8),
    ("Job D", 1, 2),
]

schedule, lateness = earliest_deadline_first(jobs)
print(f"Schedule: {schedule}")
print(f"Max lateness: {lateness}")

Shortest Processing Time (SPT)

Minimizes average completion time.

def shortest_processing_time(jobs):
    """
    Schedule jobs by shortest duration first.
    
    Args:
        jobs: list of (name, duration, deadline)
    
    Returns:
        schedule: ordered list of job names
        avg_completion: average completion time
    """
    sorted_jobs = sorted(jobs, key=lambda j: j[1])  # Sort by duration
    schedule = []
    current_time = 0
    total_completion = 0
    
    for name, duration, deadline in sorted_jobs:
        schedule.append(name)
        current_time += duration
        total_completion += current_time
    
    avg_completion = total_completion / len(jobs)
    return schedule, avg_completion


# Example
schedule, avg = shortest_processing_time(jobs)
print(f"SPT Schedule: {schedule}")
print(f"Avg completion: {avg}")

FastAPI Scheduling API

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI(title="Scheduling API")

class Job(BaseModel):
    name: str
    duration: float
    deadline: float

class ScheduleResponse(BaseModel):
    algorithm: str
    schedule: List[str]
    metric: float
    description: str

@app.post("/schedule/edf", response_model=ScheduleResponse)
async def schedule_edf(jobs: List[Job]):
    """Schedule jobs by Earliest Deadline First."""
    if not jobs:
        raise HTTPException(status_code=400, detail="At least one job required")
    
    job_list = [(j.name, j.duration, j.deadline) for j in jobs]
    schedule, max_lateness = earliest_deadline_first(job_list)
    
    return ScheduleResponse(
        algorithm="Earliest Deadline First",
        schedule=schedule,
        metric=max_lateness,
        description=f"Max lateness: {max_lateness} time units"
    )

@app.post("/schedule/spt", response_model=ScheduleResponse)
async def schedule_spt(jobs: List[Job]):
    """Schedule jobs by Shortest Processing Time."""
    if not jobs:
        raise HTTPException(status_code=400, detail="At least one job required")
    
    job_list = [(j.name, j.duration, j.deadline) for j in jobs]
    schedule, avg_completion = shortest_processing_time(job_list)
    
    return ScheduleResponse(
        algorithm="Shortest Processing Time",
        schedule=schedule,
        metric=avg_completion,
        description=f"Average completion time: {avg_completion:.2f} time units"
    )

Running the API

# Install
pip install fastapi uvicorn

# Run
uvicorn main:app --reload --port 8000

# Test
curl -X POST http://localhost:8000/schedule/edf \
  -H "Content-Type: application/json" \
  -d '[
    {"name": "Job A", "duration": 3, "deadline": 5},
    {"name": "Job B", "duration": 2, "deadline": 3},
    {"name": "Job C", "duration": 4, "deadline": 8}
  ]'

When Each Algorithm Is Best

| Algorithm | Best For | Limitation | |-----------|----------|------------| | EDF | Minimizing max lateness | May miss many deadlines slightly | | SPT | Minimizing avg completion | Can miss deadlines badly | | Round Robin | Fairness, interactive systems | Poor for batch jobs | | Priority | Critical tasks first | Starvation of low-priority tasks |

Summary

Greedy scheduling algorithms are simple, fast, and optimal for specific objectives. EDF minimizes max lateness. SPT minimizes average completion time. The FastAPI Scheduling API lets you compare different scheduling strategies on your data.

Key takeaways:

EDF: sort by deadline, minimize max lateness (optimal for this objective)
SPT: sort by duration, minimize average completion (optimal for this objective)
Scheduling API: FastAPI + greedy algorithms + Pydantic models
Each algorithm optimizes a different metric — choose the right one
Greedy scheduling is optimal for single-machine scheduling with a single objective
Real scheduling (multi-machine, precedence constraints) needs more advanced methods

What's Next: Algorithm — Vehicle Routing Problem

The next course covers the Vehicle Routing Problem (VRP) — optimizing routes for fleets of vehicles with capacity and time window constraints.