Utilities

Standalone helpers: link extraction, file listing, the concurrent URL-checking pipeline, GitHub Actions environment detection, pyproject.toml config loading, and the CLI spinner.

GitHub Environment

Module to determine the GitHub repository blob URL when running inside GitHub Actions.

markdown_checker.utils.github_env.get_github_repo_blob_url()

Build the <server>/<owner>/<repo>/blob/<sha> URL for the commit currently being checked, inferring the correct source repository from the GitHub Actions event context.

For pull_request/pull_request_target events, the source (head) repository and commit are used so links point at the contributor’s fork/branch rather than the base repo’s merge commit. For all other events (e.g. push), the repository and commit that triggered the workflow are used.

Returns:

The blob base URL, or None if not running in GitHub Actions or the required information could not be determined.

Return type:

str | None

List Files

This module contains a function to get a list of file paths from a root directory and its subdirectories, filtered by file extension.

markdown_checker.utils.list_files.get_files_paths_list(root_path, extensions=None)

Get a list of file paths from a root directory and its subdirectories, filtered by file extension.

Parameters:

  • root_path (Path) – The root directory to start the search.

  • extensions (list[str]) – A list of file extensions to filter the search.

Returns:

A (sub_folders, files_paths) tuple – all subdirectories found (recursively, regardless of extensions), and all files matching extensions (recursively).

Return type:

tuple[list[Path], list[Path]]

Spinner

class markdown_checker.utils.spinner.Spinner(beep=False, disable=False, force=False, stream=sys.stdout)

Bases: object

A simple text spinner shown on a stream while a long-running task runs.

Use via the spinner() factory / context manager rather than constructing directly. The spinner runs on a background thread and only animates when the target stream is a TTY (or force=True); otherwise start() is a no-op so redirected output (e.g. in CI logs) stays clean.

Parameters:
spinner_cycle
start()

Start animating on a background thread, unless disabled or non-TTY (see force).

Return type:

None

stop()

Signal the background thread to stop and wait for it to finish.

Return type:

None

init_spin()

Background-thread target: write "Checking... " then animate until stop() is called.

Return type:

None

markdown_checker.utils.spinner.spinner(beep=False, disable=False, force=False, stream=sys.stdout)

This function creates a context manager that is used to display a spinner on stdout as long as the context has not exited.

The spinner is created only if stdout is not redirected, or if the spinner is forced using the force parameter.

Parameters:

  • beep (bool) – Beep when spinner finishes.

  • disable (bool) – Hide spinner.

  • force (bool) – Force creation of spinner even when stdout is redirected.

  • stream (TextIO | Any)

Return type:

Spinner

Example

with spinner():

do_something() do_something_else()

URL Pipeline

Run-scoped URL checking pipeline.

Provides a single URLCheckService shared across all files in a run that:

  • Checks each unique normalized URL exactly once (in-memory memo), and shares the in-flight result with any duplicate occurrences submitted concurrently.

  • Serializes requests to the same host with a politeness delay between them.

  • Applies a host-level circuit breaker: a rate-limited response pauses the whole host until its Retry-After window elapses, instead of every URL on that host independently discovering and retrying the rate limit. A second consecutive rate limit for the same host is treated as persistent throttling: every other job already queued for that host is resolved as rate_limited without spending a network request on each, and the host is blocked from new dispatches until the cooldown elapses.

  • Bounds memory via backpressure: submit() blocks once max_pending jobs are in flight, so memory stays flat regardless of repo size.

submit() must only be called from a single producer thread for the lifetime of a service instance; a second calling thread raises RuntimeError.

markdown_checker.utils.url_pipeline.normalize_url(link)

Normalize a URL into a dedupe key.

Lowercases the scheme and host, strips a trailing slash from the path, and drops the fragment. Query strings are preserved as-is.

Parameters:

link (str) – The raw URL to normalize.

Returns:

The normalized URL string used as the memo/dedupe key.

Return type:

str

class markdown_checker.utils.url_pipeline.URLCheckService(config)

Bases: object

Run-scoped URL checking: dedupe memo, per-host pacing, host circuit breaker.

Parameters:

config (Config)

submit(url)

Submit a URL for checking, deduped against the run-wide memo.

Returns a Future that resolves once the underlying unique URL has been checked. Duplicate submissions of the same normalized URL, whether already resolved or still in flight, share the same result without triggering additional network activity.

Must only ever be called from a single thread for the lifetime of a service instance; calling it from more than one thread raises RuntimeError, since the memo/backpressure handshake below is not safe under concurrent producers.

Parameters:

url (MarkdownURL)

Return type:

Future[URLCheckResult]

close()

Shut down worker threads. Call only after all submitted Futures have resolved.

Return type:

None