groupby Then Aggregate With defaultdict (Python)
Pure stdlib group-then-aggregate: defaultdict(list) for the grouping pass, then a tiny per-group reducer. The version I reach for before importing pandas, plus the multi-stat variant.
By @elenamuller
February 8, 2026
·
Updated August 19, 2026
283 views
6
4.4 (8)
from __future__ import annotations
from collections import defaultdict
from statistics import mean
# I never use itertools.groupby for this case. itertools.groupby requires the
# input to be PRE-SORTED by the key, and silently produces wrong groups when
# it is not. defaultdict(list) handles arbitrary input order.
events = [
{'team': 'platform', 'engineer': 'alice', 'tickets': 7},
{'team': 'growth', 'engineer': 'bob', 'tickets': 4},
{'team': 'platform', 'engineer': 'carol', 'tickets': 9},
{'team': 'growth', 'engineer': 'dan', 'tickets': 2},
{'team': 'platform', 'engineer': 'eve', 'tickets': 5},
]
buckets: dict[str, list[dict]] = defaultdict(list)
for row in events:
buckets[row['team']].append(row)
# Now reduce each group however you want.
for team, rows in buckets.items():
total = sum(r['tickets'] for r in rows)
avg = round(mean(r['tickets'] for r in rows), 1)
print(f'{team}: {len(rows)} engineers, {total} tickets, avg {avg}')Two passes: the first builds defaultdict(list) keyed by group, the second walks each group and emits whatever aggregate you need. The reason I prefer this over itertools.groupby is that groupby requires the input to be sorted by the key and silently produces broken results otherwise; defaultdict does not care about input order. The from __future__ import annotations at the top is what lets the dict[str, list[dict]] type hint parse on Python 3.8 (the playground's Python). For datasets up to a few hundred thousand rows this is faster than constructing a DataFrame and lets the aggregation logic stay in plain Python.
from __future__ import annotations
from collections import defaultdict
from typing import Callable, Hashable, Iterable, TypeVar
T = TypeVar('T')
U = TypeVar('U')
def group_by_then(
rows: Iterable[T],
key: Callable[[T], Hashable],
agg: Callable[[list], U],
) -> dict:
buckets: dict = defaultdict(list)
for row in rows:
buckets[key(row)].append(row)
return {k: agg(v) for k, v in buckets.items()}
events = [
{'team': 'platform', 'tickets': 7},
{'team': 'growth', 'tickets': 4},
{'team': 'platform', 'tickets': 9},
{'team': 'growth', 'tickets': 2},
{'team': 'platform', 'tickets': 5},
]
# count per team
print(group_by_then(events, key=lambda r: r['team'], agg=len))
# total tickets per team
print(group_by_then(events, key=lambda r: r['team'], agg=lambda g: sum(r['tickets'] for r in g)))
# pluck full row list per team
print(group_by_then(events, key=lambda r: r['team'], agg=list))The helper is two functions worth of code wrapped in a parameterized agg: the caller picks how to reduce each group. Splitting key and agg is what makes the helper composable: agg=len counts, agg=sum totals, agg=list plucks the rows themselves, and any custom reducer fits the same hole. I keep this in a utils/group.py module on every Python project; it is short enough that it never needs to grow into pandas. The annotations use dict and list as bare generics, so the from __future__ import annotations is mandatory on Python 3.8 to keep the file importable.
from __future__ import annotations
from collections import defaultdict
from statistics import mean, median
# When you want N stats per group at once (count, sum, avg, p50, max), do not
# walk the group N times. One pass, accumulator pattern.
requests = [
{'route': '/api/users', 'ms': 42},
{'route': '/api/users', 'ms': 88},
{'route': '/api/cart', 'ms': 210},
{'route': '/api/users', 'ms': 51},
{'route': '/api/cart', 'ms': 195},
{'route': '/api/cart', 'ms': 310},
]
groups: dict[str, list[float]] = defaultdict(list)
for r in requests:
groups[r['route']].append(r['ms'])
report = {}
for route, samples in groups.items():
report[route] = {
'count': len(samples),
'sum': sum(samples),
'avg': round(mean(samples), 1),
'p50': round(median(samples), 1),
'max': max(samples),
}
for route, stats in sorted(report.items()):
print(route, stats)This is the shape of every per-route latency report I have shipped. The aggregation is a single loop that builds per-group lists, then one comprehension that emits the stats dict; the only reason to add a second pass is if memory pressure forces you to track running stats incrementally. statistics.median is the right p50 implementation in stdlib; for percentiles other than 50 you want statistics.quantiles(samples, n=100)[k-1] for the kth percentile. I deliberately do not show p99 here because it requires statistics.quantiles which is Python 3.8+ but with subtle behavior on tiny samples; mention it in your real code's docstring so a junior does not call it on a 5-element list.