Libraries to remember for when they may be needed

[joocer]

Fast Base64 encoding/decoding https://github.com/mayeut/pybase64
For key handling in Hadro

Radix Trees https://github.com/blackwithwhite666/pyart
For indexing text (for Hadro or Full Text Index)

https://github.com/pydata/numexpr - potentially faster numeric expressions

https://github.com/pyparsing/pyparsing/ - improve temporal extraction

https://github.com/ashvardanian/StringZilla - potentially faster string functions

https://github.com/unum-cloud/usearch - vector index

Negative inputs
https://github.com/minimaxir/big-list-of-naughty-strings

https://github.com/loxxous/Behemoth-Rank-Coding - string compression

[joocer]

Flow control and Rate Limiting (Token Bucket)

import time
import threading

class TokenBucket:
    """
    Token Bucket for throttling access to a shared resource.

    Parameters:
        capacity: int
            The maximum number of tokens the bucket can hold.
        rate: float
            The rate at which tokens are added per second.
    """
    
    def __init__(self, capacity: int, rate: float):
        self.capacity = capacity
        self.rate = rate
        self.tokens = capacity
        self.lock = threading.Lock()
        
        # Start a refill thread to add tokens at the specified rate
        self.refill_thread = threading.Thread(target=self._refill, daemon=True)
        self.refill_thread.start()

    def _refill(self):
        """Refills tokens at a constant rate."""
        while True:
            with self.lock:
                # Add tokens based on the rate
                if self.tokens < self.capacity:
                    self.tokens = min(self.capacity, self.tokens + self.rate)
            time.sleep(1)

    def consume(self, tokens: int = 1) -> bool:
        """
        Attempts to consume the requested number of tokens.

        Parameters:
            tokens: int, optional
                The number of tokens to consume. Defaults to 1.

        Returns:
            bool: True if tokens were consumed, False if not enough tokens.
        """
        with self.lock:
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            else:
                return False

def worker(name: str, bucket: TokenBucket, delay: float):
    """Worker function that attempts to consume tokens at intervals."""
    while True:
        if bucket.consume():
            print(f"{name} consumed a token.")
        else:
            print(f"{name} waiting for tokens.")
        time.sleep(delay)

# Example usage
if __name__ == "__main__":
    bucket = TokenBucket(capacity=5, rate=1)  # Max 5 tokens, refills 1 token per second

    # Create two worker threads simulating independent processes
    worker1 = threading.Thread(target=worker, args=("Worker 1", bucket, 0.5))
    worker2 = threading.Thread(target=worker, args=("Worker 2", bucket, 1.0))

    # Start the workers
    worker1.start()
    worker2.start()

    # Keep main thread alive
    worker1.join()
    worker2.join()