Initial commit

new file mode 100644

@@ -0,0 +1,20 @@

import requests

import time

def call_api_with_retry(url, method='GET', headers=None, data=None, max_retries=3):

    \"\"\"

    Simple template for calling an API with basic retry logic.

    \"\"\"

    for attempt in range(max_retries):

        try:

            response = requests.request(method, url, headers=headers, json=data)

            response.raise_for_status()

            return response.json()

        except requests.exceptions.RequestException as e:

            if attempt == max_retries - 1:

                raise e

            time.sleep(2 ** attempt)  # Exponential backoff

    return None

if __name__ == \"__main__\":\n    # Example usage:\n    # try:\n    #     data = call_api_with_retry(\"https://api.example.com/data\")\n    #     print(data)\n    # except Exception as e:\n    #     print(f\"API call failed: {e}\")

    pass

new file mode 100644

@@ -0,0 +1,32 @@

import json

import os

def load_config(path, default=None):

    """

    Safely load a JSON configuration file.

    Returns default if file does not exist or is invalid.

    """

    if not os.path.exists(path):

        return default or {}

    try:

        with open(path, 'r') as f:

            return json.load(f)

    except (json.JSONDecodeError, IOError):

        return default or {}

def save_config(path, data):

    """

    Save data to a JSON configuration file with pretty printing.

    Creates parent directories if they don't exist.

    """

    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)

    with open(path, 'w') as f:

        json.dump(data, f, indent=4)

if __name__ == "__main__":

    test_path = "test_config.json"

    data = {"version": "1.0", "status": "active"}

    save_config(test_path, data)

    loaded = load_config(test_path)

    print(f"Loaded: {loaded}")

    os.remove(test_path)

new file mode 100644

@@ -0,0 +1,18 @@

import time

def exponential_backoff(retries, base_delay=1, max_delay=60):

    """

    Returns the delay for the current retry attempt using exponential backoff.

    :param retries: Current retry attempt number (0-indexed)

    :param base_delay: Initial delay in seconds

    :param max_delay: Maximum delay in seconds

    """

    delay = min(max_delay, base_delay * (2 ** retries))

    return delay

# Example usage:

# for i in range(5):

#     wait = exponential_backoff(i)

#     print(f"Retry {i+1}: Waiting {wait}s...")

#     time.sleep(wait)

new file mode 100644

@@ -0,0 +1,28 @@

import json

import logging

def safe_json_parse(json_string, default=None):

    """

    Safely parses a JSON string, returning a default value if parsing fails.

    Args:

        json_string: The string to parse.

        default: The value to return if parsing fails (defaults to None).

    Returns:

        The parsed dictionary/list or the default value.

    """

    try:

        if not json_string:

            return default

        return json.loads(json_string)

    except (json.JSONDecodeError, TypeError) as e:

        logging.error(f"Failed to parse JSON: {e}")

        return default

# Example usage:

# data = safe_json_parse('{"key": "value"}', default={})

# print(data["key"]) # value

# 

# bad_data = safe_json_parse('invalid json', default={})

# print(bad_data) # {}

new file mode 100644

@@ -0,0 +1,25 @@

import time

def rate_limiter(calls_per_second):

    """

    A simple generator-based rate limiter.

    Yields when the next call is allowed.

    """

    interval = 1.0 / calls_per_second

    next_call = time.time()

    while True:

        now = time.time()

        if now < next_call:

            time.sleep(next_call - now)

            now = time.time()

        yield

        next_call = now + interval

if __name__ == "__main__":

    # Example usage: limit to 2 calls per second

    limiter = rate_limiter(2)

    start = time.time()

    for i in range(5):

        next(limiter)

        print(f"Call {i} at {time.time() - start:.2f}s")

new file mode 100644

@@ -0,0 +1,37 @@

import time

import random

def retry_with_exponential_backoff(func, max_retries=5, base_delay=1, max_delay=32):

    """

    Retries a function with exponential backoff.

    Args:

        func: The function to retry. Must return a truthy value on success or raise an exception on failure.

        max_retries: Maximum number of retries.

        base_delay: Initial delay in seconds.

        max_delay: Maximum delay in seconds.

    """

    retries = 0

    while retries < max_retries:

        try:

            result = func()

            if result:

                return result

        except Exception as e:

            print(f"Attempt {retries + 1} failed: {e}")

        delay = min(base_delay * (2 ** retries) + random.uniform(0, 1), max_delay)

        print(f"Retrying in {delay:.2f} seconds...")

        time.sleep(delay)

        retries += 1

    raise Exception(f"Failed after {max_retries} retries.")

# Example usage:

# def my_unreliable_api_call():

#     if random.random() < 0.7:

#         raise Exception("API Error")

#     return "Success"

# 

# result = retry_with_exponential_backoff(my_unreliable_api_call)

# print(result)