Zero-Dependency Python Recipes

import urllib.request
import urllib.error
import json

def http_json_request(url: str, method: str = 'GET', payload: dict = None, timeout: float = 10.0) -> dict:
    headers = {'User-Agent': 'Awesome-Stdlib/1.0', 'Accept': 'application/json'}
    data = json.dumps(payload).encode('utf-8') if payload is not None else None
    if data:
        headers['Content-Type'] = 'application/json'
    req = urllib.request.Request(url, data=data, headers=headers, method=method.upper())
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode('utf-8'))

import graphlib

def compute_execution_order(dag: dict[str, set[str]]) -> list[str]:
    ts = graphlib.TopologicalSorter(dag)
    return list(ts.static_order())

import hmac
import hashlib
import json

def sign_json_receipt(secret_key: bytes, payload: dict) -> str:
    encoded_bytes = json.dumps(payload, separators=(',', ':'), sort_keys=True, ensure_ascii=False, allow_nan=False).encode('utf-8')
    return hmac.new(secret_key, encoded_bytes, hashlib.sha256).hexdigest()

def verify_json_receipt(secret_key: bytes, payload: dict, expected_sig: str) -> bool:
    candidate_sig = sign_json_receipt(secret_key, payload)
    return hmac.compare_digest(expected_sig.encode(), candidate_sig.encode())

import time
from threading import Lock

class TokenBucketLimiter:
    def __init__(self, refill_rate_per_sec: float, max_tokens: int):
        self.rate = refill_rate_per_sec
        self.capacity = max_tokens
        self.tokens = float(max_tokens)
        self.last_refill = time.monotonic()
        self.lock = Lock()

    def acquire(self, tokens: int = 1) -> bool:
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.last_refill = now
            self.tokens = min(float(self.capacity), self.tokens + elapsed * self.rate)
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

import hashlib
import hmac
import secrets

def hash_password_scrypt(password: str, salt: bytes = None) -> tuple[bytes, bytes]:
    if salt is None:
        salt = secrets.token_bytes(16)
    hashed = hashlib.scrypt(password.encode('utf-8'), salt=salt, n=16384, r=8, p=1, maxmem=32 * 1024 * 1024)
    return hashed, salt

def verify_password_scrypt(password: str, hashed: bytes, salt: bytes) -> bool:
    candidate = hashlib.scrypt(password.encode('utf-8'), salt=salt, n=16384, r=8, p=1, maxmem=32 * 1024 * 1024)
    return hmac.compare_digest(candidate, hashed)

import concurrent.futures
from typing import Callable, Any, Iterable

class BoundedTaskPool:
    def __init__(self, max_workers: int = 4):
        self.max_workers = max_workers
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)

    def map(self, func: Callable[[Any], Any], iterable: Iterable[Any]) -> list[Any]:
        futures = [self.executor.submit(func, item) for item in iterable]
        return [f.result() for f in futures]

    def shutdown(self, wait: bool = True) -> None:
        self.executor.shutdown(wait=wait)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.shutdown(wait=True)

from typing import List, Dict, Any

def render_table(rows: List[Dict[str, Any]], headers: List[str]) -> str:
    if not rows:
        return ''
    widths = {h: max(len(h), max(len(str(r.get(h, ''))) for r in rows)) for h in headers}
    header_line = '| ' + ' | '.join(f"{h:<{widths[h]}}" for h in headers) + ' |'
    divider_line = '+-' + '-+-'.join('-' * widths[h] for h in headers) + '-+'
    data_lines = [
        '| ' + ' | '.join(f"{str(r.get(h, '')):<{widths[h]}}" for h in headers) + ' |'
        for r in rows
    ]
    return f"{divider_line}\n{header_line}\n{divider_line}\n" + '\n'.join(data_lines) + f"\n{divider_line}"

from dataclasses import dataclass, asdict, field
from typing import Any, Dict, Optional
import json

@dataclass(frozen=True)
class DataContract:
    contract_id: str
    max_budget_usd: float
    allowed_ops: list[str] = field(default_factory=list)
    metadata: Optional[Dict[str, Any]] = None

    def __post_init__(self):
        if self.max_budget_usd <= 0:
            raise ValueError('max_budget_usd must be positive')
        if not self.contract_id.strip():
            raise ValueError('contract_id cannot be empty')

    def to_json(self) -> str:
        return json.dumps(asdict(self), separators=(',', ':'), sort_keys=True)

import mmap
import os

def create_shared_kill_switch(path: str) -> mmap.mmap:
    with open(path, 'a+b') as f:
        if os.path.getsize(path) < 1:
            f.write(b'\x00')
            f.flush()
        return mmap.mmap(f.fileno(), 1, access=mmap.ACCESS_WRITE)

def trip_kill_switch(shm: mmap.mmap):
    shm[0] = 1

def is_kill_switch_tripped(shm: mmap.mmap) -> bool:
    return shm[0] == 1

import tomllib
from pathlib import Path
from typing import Dict, Any

def read_toml_config(file_path: Path) -> Dict[str, Any]:
    with open(file_path, 'rb') as f:
        return tomllib.load(f)

import time
import random
import functools
from typing import Callable, Type, Tuple, Any

def retry_with_backoff(retries: int = 3, base_delay: float = 0.5, max_delay: float = 10.0, jitter: bool = True, retry_on: Tuple[Type[Exception], ...] = (Exception,)) -> Callable:
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            delay = base_delay
            last_err = None
            for attempt in range(retries):
                try:
                    return func(*args, **kwargs)
                except retry_on as err:
                    last_err = err
                    if attempt == retries - 1:
                        raise
                    sleep_time = min(delay, max_delay)
                    if jitter:
                        sleep_time = random.uniform(0, sleep_time)
                    time.sleep(sleep_time)
                    delay *= 2
            if last_err:
                raise last_err
        return wrapper
    return decorator

import sqlite3
import time
import json
from typing import Optional, Any

class SQLiteKVStore:
    def __init__(self, db_path: str = ':memory:'):
        self.conn = sqlite3.connect(db_path)
        with self.conn:
            self.conn.execute('CREATE TABLE IF NOT EXISTS kv (key TEXT PRIMARY KEY, val TEXT, expires_at REAL)')

    def set(self, key: str, value: Any, ttl_seconds: Optional[float] = None) -> None:
        expires_at = time.time() + ttl_seconds if ttl_seconds is not None else None
        serialized = json.dumps(value)
        with self.conn:
            self.conn.execute('INSERT OR REPLACE INTO kv (key, val, expires_at) VALUES (?, ?, ?)', (key, serialized, expires_at))

    def get(self, key: str) -> Optional[Any]:
        cur = self.conn.cursor()
        cur.execute('SELECT val, expires_at FROM kv WHERE key = ?', (key,))
        row = cur.fetchone()
        if not row:
            return None
        val, expires_at = row
        if expires_at and time.time() > expires_at:
            with self.conn:
                self.conn.execute('DELETE FROM kv WHERE key = ?', (key,))
            return None
        return json.loads(val)

import asyncio
from typing import Coroutine, Any, List

async def run_concurrent_tasks(coroutines: List[Coroutine]) -> List[Any]:
    results = [None] * len(coroutines)
    async def _wrap(idx: int, coro: Coroutine):
        results[idx] = await coro

    async with asyncio.TaskGroup() as tg:
        for idx, coro in enumerate(coroutines):
            tg.create_task(_wrap(idx, coro))
    return results

import logging
import json
import datetime

class JSONLogFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            'timestamp': datetime.datetime.fromtimestamp(record.created, datetime.timezone.utc).isoformat(),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
        }
        if record.exc_info:
            log_entry['exception'] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

import tempfile
import os
from pathlib import Path

def write_file_atomic(target_path: Path, content: str, encoding: str = 'utf-8') -> None:
    target_path = Path(target_path)
    parent = target_path.parent
    parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile('w', dir=parent, delete=False, encoding=encoding) as tf:
        tf.write(content)
        tf.flush()
        os.fsync(tf.fileno())
        temp_name = tf.name
    os.replace(temp_name, target_path)

import secrets
import string

def generate_secure_token(length: int = 32) -> str:
    return secrets.token_urlsafe(length)[:length]

def generate_alphanumeric_code(length: int = 16) -> str:
    alphabet = string.ascii_letters + string.digits
    return ''.join(secrets.choice(alphabet) for _ in range(length))

import csv
import io
from typing import Iterator, Dict

def stream_csv_dicts(csv_data: str) -> Iterator[Dict[str, str]]:
    f = io.StringIO(csv_data)
    reader = csv.DictReader(f)
    for row in reader:
        yield dict(row)

import functools
from typing import Callable

def memoize_lru(maxsize: int = 128) -> Callable:
    def decorator(func: Callable) -> Callable:
        return functools.lru_cache(maxsize=maxsize)(func)
    return decorator

import argparse
from typing import Callable, Dict

class CLIDispatcher:
    def __init__(self, description: str):
        self.parser = argparse.ArgumentParser(description=description)
        self.subparsers = self.parser.add_subparsers(dest='command', required=True)
        self.handlers: Dict[str, Callable] = {}

    def register(self, name: str, help_text: str, func: Callable) -> argparse.ArgumentParser:
        sub = self.subparsers.add_parser(name, help=help_text)
        self.handlers[name] = func
        return sub

    def execute(self, argv: list[str] = None) -> int:
        args = self.parser.parse_args(argv)
        handler = self.handlers.get(args.command)
        return handler(args) if handler else 1

import collections
import time
from threading import Lock

class SlidingWindowCounter:
    def __init__(self, window_seconds: float, max_hits: int):
        self.window = window_seconds
        self.max_hits = max_hits
        self.hits = collections.deque()
        self.lock = Lock()

    def record_and_check(self) -> bool:
        with self.lock:
            now = time.monotonic()
            cutoff = now - self.window
            while self.hits and self.hits[0] <= cutoff:
                self.hits.popleft()
            if len(self.hits) < self.max_hits:
                self.hits.append(now)
                return True
            return False

import datetime

def cron_values(pattern: str, minimum: int, maximum: int) -> set[int]:
    values = set()
    for item in pattern.split(','):
        base, separator, step_text = item.partition('/')
        step = int(step_text) if separator else 1
        if step <= 0:
            raise ValueError('cron step must be positive')
        if base == '*':
            start, end = minimum, maximum
        elif '-' in base:
            start_text, end_text = base.split('-')
            start, end = int(start_text), int(end_text)
        else:
            if separator:
                raise ValueError('steps require a wildcard or range')
            start = end = int(base)
        if not minimum <= start <= end <= maximum:
            raise ValueError('cron value outside field range')
        values.update(range(start, end + 1, step))
    return values

def is_cron_due(cron_expr: str, dt: datetime.datetime) -> bool:
    minute_p, hour_p, day_p, month_p, dow_p = cron_expr.split()
    minutes = cron_values(minute_p, 0, 59)
    hours = cron_values(hour_p, 0, 23)
    days = cron_values(day_p, 1, 31)
    months = cron_values(month_p, 1, 12)
    weekdays = {day % 7 for day in cron_values(dow_p, 0, 7)}
    day_matches = dt.day in days
    weekday_matches = (dt.weekday() + 1) % 7 in weekdays
    if day_p.startswith('*') or dow_p.startswith('*'):
        calendar_matches = day_matches and weekday_matches
    else:
        calendar_matches = day_matches or weekday_matches
    return (dt.minute in minutes and dt.hour in hours
            and dt.month in months and calendar_matches)

from collections import defaultdict
from threading import Lock
from typing import Callable, Any

class EventEmitter:
    def __init__(self):
        self._listeners = defaultdict(list)
        self._lock = Lock()

    def on(self, event: str, listener: Callable[[Any], None]) -> None:
        with self._lock:
            self._listeners[event].append(listener)

    def emit(self, event: str, *args, **kwargs) -> None:
        with self._lock:
            callbacks = list(self._listeners.get(event, []))
        for cb in callbacks:
            cb(*args, **kwargs)

import ipaddress

def is_ip_in_network(ip_str: str, cidr_str: str) -> bool:
    ip = ipaddress.ip_address(ip_str)
    net = ipaddress.ip_network(cidr_str, strict=False)
    return ip in net

def is_private_ip(ip_str: str) -> bool:
    return ipaddress.ip_address(ip_str).is_private

import time
import secrets
import uuid

_last_ms = 0
_counter = 0

def generate_uuidv7() -> uuid.UUID:
    global _last_ms, _counter
    now_ms = int(time.time() * 1000)
    if now_ms <= _last_ms:
        now_ms = _last_ms
        _counter = (_counter + 1) & 0xFFF
        if _counter == 0:
            now_ms += 1
            _last_ms = now_ms
    else:
        _last_ms = now_ms
        _counter = 0
    ts_bytes = now_ms.to_bytes(6, byteorder='big')
    rand_bytes = secrets.token_bytes(8)
    octet6 = 0x70 | ((_counter >> 8) & 0x0F)
    octet7 = _counter & 0xFF
    octet8 = 0x80 | (rand_bytes[0] & 0x3F)
    raw = ts_bytes + bytes([octet6, octet7, octet8]) + rand_bytes[1:]
    return uuid.UUID(bytes=raw)

import hashlib
from typing import BinaryIO

def hash_binary_stream(stream: BinaryIO) -> str:
    if hasattr(hashlib, 'file_digest'):
        return hashlib.file_digest(stream, 'sha256').hexdigest()
    hasher = hashlib.sha256()
    while chunk := stream.read(65536):
        hasher.update(chunk)
    return hasher.hexdigest()

import os
from typing import TypeVar, Type, Optional

T = TypeVar('T', str, int, float, bool)

def get_env_var(name: str, default: Optional[T] = None, var_type: Type[T] = str) -> T:
    val = os.environ.get(name)
    if val is None:
        if default is not None:
            return default
        raise KeyError(f"Missing required environment variable '{name}'")
    if var_type is bool:
        return val.lower() in ('true', '1', 'yes', 'on')
    return var_type(val)

import http.server
import socketserver
import threading

class HealthHandler(http.server.SimpleHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/health':
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            self.wfile.write(b'{"status":"ok"}')
        else:
            super().do_GET()

def start_server_in_thread(port: int = 8080) -> socketserver.TCPServer:
    server = socketserver.TCPServer(('', port), HealthHandler)
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()
    return server

import heapq
import threading
from dataclasses import dataclass, field
from typing import Any, Optional

@dataclass(order=True)
class PrioritizedTask:
    priority: int
    data: Any = field(compare=False)

class PriorityQueue:
    def __init__(self):
        self._heap = []
        self._lock = threading.Lock()

    def push(self, item: Any, priority: int = 100) -> None:
        with self._lock:
            heapq.heappush(self._heap, PrioritizedTask(priority, item))

    def pop(self) -> Optional[Any]:
        with self._lock:
            if not self._heap:
                return None
            return heapq.heappop(self._heap).data

    def __len__(self) -> int:
        with self._lock:
            return len(self._heap)

from pathlib import Path
from typing import Iterator

def walk_files_with_extension(root_dir: Path, extension: str) -> Iterator[Path]:
    root = Path(root_dir)
    ext = extension if extension.startswith('.') else f".{extension}"
    for p in root.rglob(f"*{ext}"):
        if p.is_file():
            yield p

import signal
import threading
from typing import Optional

class ShutdownCoordinator:
    def __init__(self):
        self._stop_event = threading.Event()
        signal.signal(signal.SIGINT, self._handle_signal)
        signal.signal(signal.SIGTERM, self._handle_signal)

    def _handle_signal(self, signum, frame):
        self._stop_event.set()

    def is_shutting_down(self) -> bool:
        return self._stop_event.is_set()

    def wait(self, timeout: Optional[float] = None) -> bool:
        return self._stop_event.wait(timeout)