Created
December 10, 2025 20:02
-
-
Save it9gamelog/d32cdad2a879914fa40778a70dabead1 to your computer and use it in GitHub Desktop.
Python utility functions for simple making failable HTTP webhook call, such as health checking heartbeat.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| Utility functions for heartbeat requests. | |
| """ | |
| import os | |
| import requests | |
| from typing import Optional, Tuple | |
| from urllib.parse import urlparse | |
| def parse_heartbeat_config(config_str: str) -> Tuple[str, str, Optional[str]]: | |
| """ | |
| Parse heartbeat configuration string. | |
| Formats supported: | |
| - METHOD|URL|bearer_token | |
| - URL|bearer_token | |
| - METHOD|URL | |
| - URL | |
| Args: | |
| config_str: The configuration string from the secret | |
| Returns: | |
| Tuple of (method, url, bearer_token) | |
| - method: HTTP method (defaults to GET if not specified) | |
| - url: The URL (must start with http:// or https://) | |
| - bearer_token: Optional bearer token (None if not present) | |
| """ | |
| parts = [p.strip() for p in config_str.split("|")] | |
| # Determine if first part is a method or URL | |
| method = "GET" # Default | |
| url = None | |
| bearer_token = None | |
| if len(parts) == 1: | |
| # Format: URL | |
| url = parts[0] | |
| elif len(parts) == 2: | |
| # Could be METHOD|URL or URL|bearer_token | |
| # Check if first part is a valid HTTP method | |
| if parts[1].startswith(("http://", "https://")): | |
| # Format: METHOD|URL | |
| method = parts[0].upper() | |
| url = parts[1] | |
| else: | |
| # Format: URL|bearer_token | |
| url = parts[0] | |
| bearer_token = parts[1] | |
| elif len(parts) == 3: | |
| # Format: METHOD|URL|bearer_token | |
| method = parts[0].upper() | |
| url = parts[1] | |
| bearer_token = parts[2] | |
| else: | |
| raise ValueError(f"Invalid heartbeat config format: {config_str}") | |
| # Validate URL | |
| if not url.startswith(("http://", "https://")): | |
| raise ValueError(f"URL must start with http:// or https://: {url}") | |
| # Validate URL format | |
| try: | |
| parsed = urlparse(url) | |
| if not parsed.netloc: | |
| raise ValueError(f"Invalid URL format: {url}") | |
| except Exception as e: | |
| raise ValueError(f"Invalid URL: {url}") from e | |
| return (method, url, bearer_token) | |
| def read_heartbeat_config(secret_path: str) -> Optional[str]: | |
| """ | |
| Read heartbeat configuration from mounted secret file. | |
| Args: | |
| secret_path: Path to the secret file | |
| Returns: | |
| The configuration string, or None if file doesn't exist | |
| """ | |
| if not os.path.exists(secret_path): | |
| return None | |
| try: | |
| with open(secret_path, "r", encoding="utf-8") as f: | |
| config = f.read().strip() | |
| return config if config else None | |
| except Exception as e: | |
| print(f"Error reading heartbeat config from {secret_path}: {e}") | |
| return None | |
| def send_heartbeat( | |
| method: str, | |
| url: str, | |
| bearer_token: Optional[str] = None, | |
| timeout: int = 5, | |
| max_retries: int = 3, | |
| ) -> bool: | |
| """ | |
| Send heartbeat request with retries. | |
| Args: | |
| method: HTTP method (GET, POST, etc.) | |
| url: The URL to send the request to | |
| bearer_token: Optional bearer token for authorization | |
| timeout: Request timeout in seconds (default: 5) | |
| max_retries: Maximum number of retries on failure (default: 3) | |
| Returns: | |
| True if request succeeded, False otherwise | |
| """ | |
| headers = {} | |
| if bearer_token: | |
| headers["Authorization"] = f"Bearer {bearer_token}" | |
| for attempt in range(max_retries): | |
| try: | |
| response = requests.request( | |
| method=method, | |
| url=url, | |
| headers=headers, | |
| timeout=timeout, | |
| ) | |
| response.raise_for_status() | |
| print(f"Heartbeat sent successfully: {method} {url} - Status: {response.status_code}") | |
| return True | |
| except requests.exceptions.RequestException as e: | |
| if attempt < max_retries - 1: | |
| print(f"Heartbeat attempt {attempt + 1}/{max_retries} failed: {e}, retrying...") | |
| else: | |
| print(f"Heartbeat failed after {max_retries} attempts: {e}") | |
| return False | |
| def send_heartbeat_from_config(secret_path: str) -> bool: | |
| """ | |
| Read heartbeat configuration and send the heartbeat request. | |
| Args: | |
| secret_path: Path to the secret file containing the configuration | |
| Returns: | |
| True if heartbeat was sent successfully, False otherwise | |
| """ | |
| config_str = read_heartbeat_config(secret_path) | |
| if not config_str: | |
| print(f"Heartbeat config not found at {secret_path}, skipping heartbeat") | |
| return False | |
| try: | |
| method, url, bearer_token = parse_heartbeat_config(config_str) | |
| return send_heartbeat(method, url, bearer_token, timeout=5, max_retries=3) | |
| except Exception as e: | |
| print(f"Failed to send heartbeat: {e}") | |
| return False | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment