Reference library

Data pipelines & processing

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

4 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 List Failed Records in a Dead Letter Queue Mock in Python

A mock Dead Letter Queue stores failed processing records with error details and timestamps, lists them, and exports to JSON.

dead-letter-queue json logging

Python

import json
from datetime import datetime, timedelta
import random


class DeadLetterQueue:
    def __init__(self):
        self.failed_records = []

    def add_failed_record(self, record_id, payload, error_message):
        self.failed_records.append({
            "record_id": record_id,
            "payload": paylo…

13 0 Open

Data pipelines & processing easy

How to Sort a List of Dictionaries by Key in Python

A reusable helper function that sorts a list of dictionaries by a specified key, with optional descending order support.

sorting dictionaries data-pipelines

Python

from typing import List

def sort_records(records: List[dict], key: str, descending: bool = False) -> List[dict]:
    """Sort a list of dictionaries by a specified key."""
    return sorted(records, key=lambda record: record[key], reverse=descending)


def demonstrate_sorting() -> None:
    users = [
        {"name": …

12 0 Open

Data pipelines & processing easy

Test a Python Pipeline with Fixture Sample Rows

Test pipeline functions with sample rows provided by a pytest fixture, verifying required keys and value constraints.

pytest fixtures data-pipelines

Python

import pytest


def get_value(data: dict, key: str):
    return data.get(key)


def sample_rows():
    return [
        {"name": "Alice", "age": 30, "city": "London"},
        {"name": "Bob", "age": 25, "city": "Paris"},
        {"name": "Charlie", "age": 35, "city": "Berlin"},
    ]


@pytest.fixture
def sample_data(…

16 0 Open

Browse by section

Each section groups closely related Python snippets.