Using context

Author: Denis-Averin

scripts/check-urls.py

@@ -20,7 +20,7 @@
 Check them with CURL
 """
 
-JOIN_TIMEOUT_SEC: int = 120
+JOIN_TIMEOUT_SEC = 120
 
 CURL_EXIT_CODES_AND_HTTP_CODES: dict[str, tuple[int, int | None]] = {
     "https://api.aspose.cloud/connect/token": (CURL_EXIT_CODES.HTTP_RETURNED_ERROR, 400),
@@ -36,15 +36,15 @@
     re.compile(r"^https://github\.com/(?P<user>[^/]+)/(?P<repo>[^/]+)/(?:blob|issues)/\S+$"),
 ]
 
-URLS_TO_IGNORE: frozenset[str] = frozenset(
+URLS_TO_IGNORE = frozenset(
     [
         "https://api.aspose.cloud",
         "https://www.aspose.cloud/404",
         "https://www.aspose.cloud/404/",
     ]
 )
 
-IGNORE_DOMAINS: Subdomains = Subdomains(
+IGNORE_DOMAINS = Subdomains(
     [
         ".android.com",
         ".apache.org",
@@ -81,10 +81,10 @@
     ]
 )
 
-URL_END_CHARS: str = r",#\)\"'<>\*\s\\"
-URL_RE_PATTERN: str = r"(https*://[^{0}]+)[{0}]?".format(URL_END_CHARS)
+URL_END_CHARS = r",#\)\"'<>\*\s\\"
+URL_RE_PATTERN = r"(https*://[^{0}]+)[{0}]?".format(URL_END_CHARS)
 # print(URL_RE_PATTERN)
-EXTRACT_URL_REGEX: re.Pattern[str] = re.compile(URL_RE_PATTERN, re.MULTILINE)
+EXTRACT_URL_REGEX = re.compile(URL_RE_PATTERN, re.MULTILINE)
 
 # URL : [Files]
 EXTRACTED_URLS_WITH_FILES: dict[str, list[str]] = {k: [] for k in URLS_TO_IGNORE}
@@ -128,7 +128,7 @@ def url_extractor(text: str, filename: str) -> typing.Generator[str, None, None]
             EXTRACTED_URLS_WITH_FILES[url].append(filename)
 
 
-FILES_TO_IGNORE: frozenset[str] = frozenset(
+FILES_TO_IGNORE = frozenset(
     [
         ".jar",
         ".jar",
@@ -153,7 +153,7 @@ def text_extractor(files: list[str]) -> typing.Generator[tuple[str, str], None,
                     raise
 
 
-JOB_SUMMARY: JobSummary = JobSummary(os.environ.get("GITHUB_STEP_SUMMARY", "step_summary.md"))
+JOB_SUMMARY = JobSummary(os.environ.get("GITHUB_STEP_SUMMARY", "step_summary.md"))
 JOB_SUMMARY.add_header("Test all URLs")
 
 
@@ -162,32 +162,15 @@ def main(files: list[str]) -> int:
         expectations=CURL_EXIT_CODES_AND_HTTP_CODES,
     )
 
-    # Setup signal handlers for graceful shutdown
-    def _handle_signal(_sig: int, _frame: typing.Any) -> None:
-        url_checker.stop()
-
-    with contextlib.suppress(Exception):
-        signal.signal(signal.SIGINT, _handle_signal)
-        signal.signal(signal.SIGTERM, _handle_signal)
-
-    checker = threading.Thread(target=url_checker.run, daemon=True)
-    checker.start()
-
-    for filename, text in text_extractor(files):
-        for url in url_extractor(text, filename):
-            # print("In:", url)
-            url_checker.add_url(url)
-    url_checker.close()
-    checker.join(timeout=JOIN_TIMEOUT_SEC)
-    if checker.is_alive():
-        print(
-            f"URL checker did not finish within {JOIN_TIMEOUT_SEC}s; exiting early.",
-            file=sys.stderr,
-            flush=True,
-        )
+    with url_checker.start() as checker:
+        for filename, text in text_extractor(files):
+            for url in url_extractor(text, filename):
+                checker.add_url(url)
+        checker.wait(JOIN_TIMEOUT_SEC)
+    results = url_checker.results
 
     # Collect results and write summary
-    for res in url_checker.results:
+    for res in results:
         if res.ok:
             JOB_SUMMARY.add_success(res.url)
         else:

scripts/url_checker.py

@@ -3,7 +3,9 @@
 import time
 from dataclasses import dataclass
 from queue import Queue, Empty
-from typing import Callable, Optional
+from typing import Callable, Optional, Iterable
+from types import TracebackType
+import threading
 
 from curl_wrapper import CurlWrapper, EXIT_CODES
 
@@ -39,12 +41,16 @@ def __init__(
         self.stop_event = False
         self.next_report_age_sec = 5
         self.results: list[CheckResult] = []
+        self._thread: threading.Thread | None = None
+        self._closed: bool = False
 
     def add_url(self, url: str) -> None:
         self.queue.put_nowait(url)
 
     def close(self) -> None:
-        self.queue.put_nowait(None)
+        if not self._closed:
+            self._closed = True
+            self.queue.put_nowait(None)
 
     def stop(self) -> None:
         self.stop_event = True
@@ -95,6 +101,46 @@ def run(self) -> None:
             time.sleep(0.2)
         print("Worker finished")
 
+    # Context management and user-friendly API
+    def start(self) -> "UrlChecker":
+        if self._thread is not None:
+            return self
+        self._thread = threading.Thread(target=self.run, daemon=True)
+        self._thread.start()
+        return self
+
+    def __enter__(self) -> "UrlChecker":
+        return self.start()
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        # Ensure we signal end of input and wait for completion
+        self.close()
+        self.wait()
+
+    def wait(self, join_timeout_sec: float | None = None) -> None:
+        # Ensure end-of-input signaled before waiting
+        self.close()
+        t = self._thread
+        if t is None:
+            return
+        if join_timeout_sec is not None:
+            t.join(timeout=join_timeout_sec)
+            if t.is_alive():
+                # Try to stop gracefully and inform user
+                self.stop()
+                print(
+                    f"URL checker did not finish within {join_timeout_sec}s; exiting early.",
+                    file=sys.stderr,
+                    flush=True,
+                )
+        else:
+            t.join()
+
     def _process_finished(self, task: CurlWrapper) -> None:
         expected_ret_code, expected_http_code = self.expectations.get(task.url, (0, None))