Reference library

Data pipelines & processing

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

5 matches

Deduplicate events by ID within a window in Python

Deduplicate event streams by ID within sliding time windows, keeping the newest occurrence per window using heaps and sets.

deduplication events heapq

Python

import heapq
from collections import defaultdict

def deduplicate_events(events, window_size):
    """Return events deduplicated by id, keeping newest within each sliding window."""
    # Index events by (timestamp, id) for deterministic ordering
    events_by_id = defaultdict(list)
    for ts, eid, *payload in events…

14 0 Open

Data pipelines & processing easy

Generate a Deterministic Hash for Deduplication in Python

Create a stable SHA-256 fingerprint from nested data and file contents to deduplicate records in a data pipeline.

hashing deduplication sha256

Python

import hashlib
import json
from pathlib import Path

def natural_key_hash(data, salt=""):
    """
    Generate a deterministic fingerprint from raw data (dict/list/str).
    Uses JSON canonical-ish serialization with sorted keys and SHA-256.
    """
    canonical = json.dumps(data, sort_keys=True, separators=(",", ":"…

14 0 Open

Data pipelines & processing easy

How to Deduplicate Events with At-Least-Once Delivery in Python

Implements an exactly-once processing pattern for at-least-once event delivery by tracking seen event IDs in a set, skipping duplicates.

deduplication idempotent event-processing

Python

seen_ids = set()

def process_event(event_id: str, payload: dict) -> dict:
    """Process an event exactly once, ignoring duplicates."""
    if event_id in seen_ids:
        return {"status": "duplicate", "event_id": event_id}
    seen_ids.add(event_id)
    return {"status": "processed", "event_id": event_id, **payloa…

13 0 Open

Data pipelines & processing easy

Idempotent Pipeline Dedupe by Record ID Set in Python

Filters records against a persistent set of seen IDs, returning only new ones and the updated set for idempotent pipeline processing.

deduplication idempotency pipelines

Python

def dedupe_records(records, seen_ids=None):
    """Return records whose id has not been seen before."""
    if seen_ids is None:
        seen_ids = set()
    unique = []
    for record in records:
        record_id = record.get("id")
        if record_id not in seen_ids:
            seen_ids.add(record_id)
           …

12 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.