Reference library

Data pipelines & processing

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

3 matches

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

How to Track Checkpoint Offset After Batch Commit in Python

A batch processor that tracks the last successfully committed offset after processing records in batches, advancing the checkpoint only when each batch commits successfully.

batch-processing checkpoint offset

Python

import json
from typing import Any


class BatchProcessor:
    """Tracks checkpoint offset after committing batches."""

    def __init__(self, batch_size: int = 3):
        self.batch_size = batch_size
        self.offset = 0  # last successfully committed offset (exclusive)
        self.total_committed = 0

    def …

12 0 Open

Data pipelines & processing medium

Normalize Timestamps to UTC DateTime in Python

Convert timestamps in multiple formats to UTC-aware datetime objects using datetime.strptime and astimezone.

datetime timezone utc

Python

from datetime import datetime, timezone

raw_timestamps = [
    "2024-01-15 14:30:00+02:00",
    "17/05/2024 09:15:00 -0500",
    "2024-03-01T22:45:00Z",
    "2024-06-20 08:00:00+09:30"
]

def parse_and_convert(ts: str) -> datetime:
    normalized_ts = ts.strip().replace("Z", "+00:00")
    formats = [
        "%Y-%m-%…

14 0 Open

Browse by section

Each section groups closely related Python snippets.