Reference library

Data pipelines & processing

ETL-style flows, batch transforms, validation, and moving data between formats.

3 matches

How to Group Data by Key in Python

Group a list of dictionaries by a specified key using a defaultdict and compute per-group averages.

from collections import defaultdict

def group_by_key(data, key):
    grouped = defaultdict(list)
    for item in data:
        grouped[item[key]].append(item)
    return dict(grouped)

if __name__ == "__main__":
    records = [
        {"name": "Alice", "dept": "Engineering", "score": 85},
        {"name": "Bob", "de…

15 0 Open

Data pipelines & processing easy

How to Implement a Sliding Window Average in Python

Compute the average of the most recent N values in a stream using a bounded deque, efficiently updating the total as new values arrive.

deque sliding-window streaming

Python

from collections import deque


class SlidingWindowAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque(maxlen=window_size)
        self.total = 0

    def add(self, value):
        if len(self.window) == self.window_size:
            self.total -= self.windo…

15 0 Open

Data pipelines & processing easy

Implement Exactly-Once Transaction Log in Python

A mock transaction log that deduplicates transaction IDs so each is recorded only once, with a dataclass for records and simple in-memory storage.

transactions deduplication dataclass

Python

from dataclasses import dataclass
from typing import Dict, Optional


@dataclass
class TxnRecord:
    txn_id: str
    status: str


class ExactlyOnceTxnLog:
    def __init__(self) -> None:
        self._log: Dict[str, TxnRecord] = {}
        self._processed_ids: set = set()

    def record(self, txn_id: str, status: s…

14 0 Open

Browse by section

Each section groups closely related Python snippets.