| import os | |
| import time | |
| import requests | |
| import threading | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| SPACE_URL = os.getenv("SPACE_URL", "http://localhost:7860") | |
| def ping_self(): | |
| """Runs in a loop to ping the health endpoint and prevent space sleeping.""" | |
| # Delay initial ping slightly to allow FastAPI server to spin up fully | |
| time.sleep(15) | |
| while True: | |
| try: | |
| url = "http://localhost:7860/health" | |
| logger.info(f"[KeepAlive] Pinging self at {url}") | |
| response = requests.get(url, timeout=10) | |
| if response.status_code == 200: | |
| logger.info("[KeepAlive] Ping successful") | |
| else: | |
| logger.warning(f"[KeepAlive] Ping returned status code {response.status_code}") | |
| except Exception as e: | |
| logger.error(f"[KeepAlive] Ping failed: {e}") | |
| # Sleep for 25 minutes (25 * 60 seconds) | |
| time.sleep(25 * 60) | |
| def start_keep_alive(): | |
| """Starts the background ping daemon thread.""" | |
| thread = threading.Thread(target=ping_self, daemon=True) | |
| thread.start() | |
| logger.info("[KeepAlive] Background ping thread started") | |