python 21 lines · 6 steps

Round-robin task distribution in Python

A cyclic iterator hands tasks to workers in turn, with a weighted variant that skews the rotation by capacity.

Explained by highlit

Intermediate round-robin iterators load-balancing defaultdict

1from itertools import cycle
2from collections import defaultdict
3 
4 
5def distribute(tasks, workers):
6    if not workers:
7        raise ValueError("at least one worker is required")
8 
9    assignments = defaultdict(list)
10    worker_pool = cycle(workers)
11 
12    for task in tasks:
13        worker = next(worker_pool)
14        assignments[worker].append(task)
15 
16    return dict(assignments)
17 
18 
19def distribute_weighted(tasks, workers):
20    expanded = [w for w in workers for _ in range(w.capacity)]
21    return distribute(tasks, expanded)

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1itertools.cycle turns any finite sequence into an endless round-robin without tracking an index yourself.
2defaultdict(list) removes the need to check for and initialize missing keys before appending.
3Weighting a round-robin can be as simple as repeating each item in the pool proportionally to its weight.

Related explainers

python

import base64
import json
from typing import Annotated, Optional

Cursor pagination in a FastAPI endpoint

pagination cursor async

Intermediate 9 steps

python

import secrets
 
from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import APIKeyHeader

API key authentication as a FastAPI dependency

authentication dependency-injection api-keys

Intermediate 8 steps

package hashring
 
import (
	"hash/crc32"

How consistent hashing works in Go

consistent-hashing load-balancing concurrency

Intermediate 8 steps

python

import hashlib
from collections import defaultdict
from pathlib import Path

Finding duplicate files by size then hash

hashing file-io deduplication

Intermediate 7 steps

python

from flask import Flask, request, g, jsonify
from flask_babel import Babel, gettext as _, format_datetime
from datetime import datetime

Per-request localization in Flask with Babel

i18n content-negotiation request-lifecycle

Intermediate 8 steps

python

from django.core.cache import cache
from django.core.cache.utils import make_template_fragment_key
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver