Reference library

Reliability & rate limiting

Retries, exponential backoff, circuit breakers, token buckets, and idempotent handlers.

4 matches

GCRA generic cell rate algorithm in Python

Mock implementation of the Generic Cell Rate Algorithm (GCRA) for traffic shaping and rate limiting.

from collections import deque
import time

class GCRA:
    def __init__(self, rate, burst):
        self.tau = burst
        self.T = rate
        self.t = 0
        self.LCT = 0

    def add_cell(self, arrival_time):
        if arrival_time <= self.t:
            return False
        arrived_early = (arrival_time - s…

14 0 Open

Reliability & rate limiting easy

How to Deduplicate Messages in Python by ID

This code consumes a mock inbox of JSON messages and deduplicates them by message ID, keeping either the first or last occurrence.

deduplication inbox json

Python

import json
from collections import OrderedDict

mock_inbox = [
    {"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
    {"id": 2, "message": "world", "timestamp": "2024-01-01T10:01:00Z"},
    {"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
    {"id": 3, "message": "test", "times…

14 0 Open

Reliability & rate limiting easy

How to Inject Random Latency for Chaos Testing in Python

Mock unreliable services by wrapping functions with a decorator that adds random network-like delays before execution.

chaos-engineering decorators latency

Python

import random
import time
from functools import wraps

def inject_latency(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        latency = random.uniform(0.1, 0.5)
        print(f"Injecting {latency:.3f}s latency...")
        time.sleep(latency)
        return func(*args, **kwargs)
    return wrapper

@inje…

12 0 Open

Reliability & rate limiting medium

Leaky Bucket Rate Limiter in Python: Smooth Burst Traffic

Implements a token-bucket-style leaky bucket rate limiter that smooths bursty traffic by draining at a fixed rate and dropping excess packets.

rate-limiting traffic-shaping simulation

Python

import time
import random


class LeakyBucket:
    def __init__(self, capacity, drain_rate):
        self.capacity = capacity
        self.drain_rate = drain_rate
        self.water = 0.0
        self.last_time = time.time()

    def allow(self, packet_size=1.0):
        now = time.time()
        elapsed = now - self.…

13 0 Open

Browse by section

Each section groups closely related Python snippets.