Added a questionable ping and changed HTTP head to get

main.py

@@ -0,0 +1,24 @@
+# import requests as r
+from flask import jsonify, Flask
+from poll_services import start_async_loop
+from mem import services
+import threading
+
+
+# Flask app to serve status
+app = Flask(__name__)
+
+
+@app.route("/")
+def status():
+    return jsonify([s.to_dict() for s in services])
+
+
+# Only run if directly running file
+if __name__ == "__main__":
+
+    t = threading.Thread(target=start_async_loop, daemon=True)
+    t.start()
+
+    # Run flask app
+    app.run(debug=True, use_reloader=False)

mem/__init__.py

@@ -0,0 +1,55 @@
+from typing import Any, Optional
+
+
+class service:
+    url: str
+    status: Optional[int]
+    online: bool
+    public: bool
+    error: Optional[str]
+    ping: Optional[int]
+
+    def __init__(self, url: str = "", public: bool = True):
+        self.url = url
+        self.public = public
+
+        self.online = False
+        self.status = None
+        self.error = None
+        self.ping = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "url": self.url,
+            "status": self.status,
+            "public": self.public,
+            "online": self.online,
+            "error": self.error,
+            "ping": self.ping,
+        }
+
+    def set_status(self, status: Optional[int]):
+        self.status = status
+
+    def set_online(self, b: bool):
+        self.online = b
+
+    def set_error(self, s: Optional[str]):
+        self.error = s
+
+    def set_ping(self, n: Optional[int]):
+        self.ping = n
+
+
+services: list[service] = [
+    service("https://git.ihatemen.uk/"),
+    service("https://plex.ihatemen.uk/"),
+    service("https://truenas.local/", False),
+    service("https://cloud.ihatemen.uk/"),
+    service("https://request.ihatemen.uk/"),
+    service("https://id.ihatemen.uk/"),
+    service("http://tautulli.local", False),
+    service("https://transmission.local", False),
+    service("https://vault.ihatemen.uk"),
+    service("https://nginx.local", False),
+]

poll_services.py

@@ -0,0 +1,48 @@
+from mem import services, service
+import httpx
+import urllib3
+import asyncio
+import time
+
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+
+async def check_service(client: httpx.AsyncClient, s: service):
+    try:
+        before = time.perf_counter()
+        r = await client.get(
+            url=s.url,
+            follow_redirects=True,
+            timeout=1,
+        )
+        after = time.perf_counter()
+        s.set_error(None)
+        s.set_online(r.status_code == 200)
+        s.set_status(r.status_code)
+        s.set_ping(int((after - before) * 1000))
+    except httpx.HTTPError as e:
+        s.set_error(str(e))
+        s.set_online(False)
+        s.set_status(None)
+        s.set_ping(None)
+
+
+def start_async_loop():
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    asyncio.run_coroutine_threadsafe(update_services(loop=loop), loop=loop)
+    loop.run_forever()
+
+
+async def update_services(loop: asyncio.AbstractEventLoop):
+    print("Starting service updates...")
+    async with httpx.AsyncClient() as public_client, httpx.AsyncClient(
+        verify=False
+    ) as local_client:
+        while True:
+            tasks = [
+                check_service(public_client if s.public else local_client, s)
+                for s in services
+            ]
+            await asyncio.gather(*tasks)
+            await asyncio.sleep(2)

test.py

@@ -1,67 +0,0 @@
-import time
-import requests
-import traceback
-import urllib3
-from urllib.parse import urlparse
-from flask import Flask, jsonify
-from concurrent.futures import ThreadPoolExecutor, as_completed
-from typing import Any
-
-# Disable warnings for self-signed certificates
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
-services = [
-    "https://truenas.local/",
-    "https://example.com/",
-    "https://git.ihatemen.uk/"
-]
-
-services_status = {}
-
-# Determine SSL verification based on hostname
-def should_verify_ssl(url: str):
-    hostname = urlparse(url).hostname
-    return not (hostname and hostname.endswith(".local"))
-
-# Function to check a single service
-def check_service(url: str) -> tuple[str, dict[str, Any]]:
-    try:
-        start = time.time()
-        r = requests.head(url, allow_redirects=True, timeout=5, verify=should_verify_ssl(url))
-        latency = int((time.time() - start) * 1000)
-        return url, {
-            "status": "up" if r.ok else "down",
-            "latency": latency,
-            "error": None
-        }
-    except requests.exceptions.RequestException as e:
-        return url, {
-            "status": "down",
-            "latency": None,
-            "error": str(e),
-            "trace": traceback.format_exc()
-        }
-
-# Background thread that checks all services in parallel
-def check_services_periodically(interval: int=5):
-    while True:
-        with ThreadPoolExecutor(max_workers=len(services)) as executor:
-            futures = [executor.submit(check_service, url) for url in services]
-            for future in as_completed(futures):
-                url, result = future.result()
-                services_status[url] = result
-        time.sleep(interval)
-
-# Start background checker
-import threading
-threading.Thread(target=check_services_periodically, daemon=True).start()
-
-# Flask app to serve status
-app = Flask(__name__)
-
-@app.route("/status")
-def status():
-    return jsonify(services_status)
-
-if __name__ == "__main__":
-    app.run(debug=True)