Skip to content

_bundles_fetch

rhiza_hooks._bundles_fetch

Load and fetch template-bundles documents into a typed result.

This module is responsible solely for obtaining a template-bundles document — from a local file, from already-fetched bytes, or from a remote GitHub repository — and returning it as a :class:BundlesDoc. Structural validation of the returned mapping lives in :mod:rhiza_hooks._bundles_validate.

BundlesDoc dataclass

Outcome of loading/parsing a template-bundles document.

data holds the parsed mapping on success and is None on failure; errors carries the failure messages (empty on success). The two are mutually exclusive, so callers branch on data is None — which also lets the type checker narrow data to dict on the success path without a cast.

Source code in rhiza_hooks/_bundles_fetch.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
@dataclass(frozen=True)
class BundlesDoc:
    """Outcome of loading/parsing a template-bundles document.

    ``data`` holds the parsed mapping on success and is ``None`` on failure;
    ``errors`` carries the failure messages (empty on success). The two are
    mutually exclusive, so callers branch on ``data is None`` — which also lets
    the type checker narrow ``data`` to ``dict`` on the success path without a
    cast.
    """

    data: dict[Any, Any] | None
    errors: list[str]

Fetcher

Bases: Protocol

The :func:fetch_remote_bundles-shaped callable a validation run obtains its document from.

The same reasoning as :class:_Opener, one layer up. check_template_bundles injects this rather than reaching for the module global, so a test supplies a fake document through the argument every caller uses instead of rebinding check_template_bundles.fetch_remote_bundles by dotted name — a rebinding that pins the wiring rather than the behaviour, and breaks on any rename.

Unprefixed, unlike :class:_Opener: this module's convention is that a leading underscore marks a helper with no caller outside its own file, and this one is named in another module's signatures.

Source code in rhiza_hooks/_bundles_fetch.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Fetcher(Protocol):
    """The :func:`fetch_remote_bundles`-shaped callable a validation run obtains its document from.

    The same reasoning as :class:`_Opener`, one layer up. ``check_template_bundles``
    injects this rather than reaching for the module global, so a test supplies a fake
    document through the argument every caller uses instead of rebinding
    ``check_template_bundles.fetch_remote_bundles`` by dotted name — a rebinding that
    pins the wiring rather than the behaviour, and breaks on any rename.

    Unprefixed, unlike :class:`_Opener`: this module's convention is that a leading
    underscore marks a helper with no caller outside its own file, and this one is
    named in another module's signatures.
    """

    def __call__(self, repo: str, branch: str, *, attempts: int, timeout: float) -> BundlesDoc:
        """Fetch the template-bundles document for ``repo``/``branch``."""
        ...

__call__(repo, branch, *, attempts, timeout)

Fetch the template-bundles document for repo/branch.

Source code in rhiza_hooks/_bundles_fetch.py 
75
76
77
def __call__(self, repo: str, branch: str, *, attempts: int, timeout: float) -> BundlesDoc:
    """Fetch the template-bundles document for ``repo``/``branch``."""
    ...

fetch_remote_bundles(repo, branch, attempts=FETCH_ATTEMPTS, backoff=FETCH_BACKOFF_SECONDS, timeout=FETCH_TIMEOUT_SECONDS, opener=urlopen)

Fetch template-bundles.yml from a remote GitHub repository.

Transient network failures (URLError/TimeoutError) are retried up to attempts times with a linear backoff, and each failed attempt is logged so CI failures are diagnosable. HTTP errors (e.g. 404) are permanent and returned immediately without retrying.

Parameters:

Name Type Description Default
repo str

GitHub repository in 'owner/repo' format

 required
branch str

Branch name

 required
attempts int

Total number of fetch attempts (initial try + retries)

 FETCH_ATTEMPTS
backoff float

Base seconds to sleep between attempts (multiplied by attempt number)

 FETCH_BACKOFF_SECONDS
timeout float

Per-request socket timeout in seconds

 FETCH_TIMEOUT_SECONDS
opener _Opener

Performs one HTTP GET; defaults to :func:urllib.request.urlopen. Only the scheme-checked URL built below is ever passed to it — tests substitute a fake instead of rebinding this module's urlopen.

 urlopen

Returns:

Name Type Description
A BundlesDoc

class:BundlesDoc with the parsed mapping on success, or errors.
Source code in rhiza_hooks/_bundles_fetch.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def fetch_remote_bundles(
    repo: str,
    branch: str,
    attempts: int = FETCH_ATTEMPTS,
    backoff: float = FETCH_BACKOFF_SECONDS,
    timeout: float = FETCH_TIMEOUT_SECONDS,
    opener: _Opener = urlopen,  # nosec B310
) -> BundlesDoc:
    """Fetch template-bundles.yml from a remote GitHub repository.

    Transient network failures (`URLError`/`TimeoutError`) are retried up to
    ``attempts`` times with a linear backoff, and each failed attempt is logged
    so CI failures are diagnosable. HTTP errors (e.g. 404) are permanent and
    returned immediately without retrying.

    Args:
        repo: GitHub repository in 'owner/repo' format
        branch: Branch name
        attempts: Total number of fetch attempts (initial try + retries)
        backoff: Base seconds to sleep between attempts (multiplied by attempt number)
        timeout: Per-request socket timeout in seconds
        opener: Performs one HTTP GET; defaults to :func:`urllib.request.urlopen`.
            Only the scheme-checked URL built below is ever passed to it — tests
            substitute a fake instead of rebinding this module's ``urlopen``.

    Returns:
        A :class:`BundlesDoc` with the parsed mapping on success, or errors.
    """
    # Construct GitHub raw content URL
    url = f"https://raw.githubusercontent.com/{repo}/{branch}/.rhiza/template-bundles.yml"

    # Validate URL scheme for security (bandit B310)
    parsed = urlparse(url)
    if parsed.scheme != "https":
        return BundlesDoc(None, [f"Invalid URL scheme: {parsed.scheme}. Only https is allowed."])

    # pragma below: equivalent mutant — the final `return BundlesDoc(None, errors)` is only
    # reached after a transient-error iteration has reassigned `errors` (success and HTTP
    # errors return early), so for attempts >= 1 this initial value is never the one returned.
    errors: list[str] = []  # pragma: no mutate
    for attempt in range(attempts):
        outcome = _fetch_once(url, timeout, repo, branch, opener)
        if isinstance(outcome, BundlesDoc):
            return outcome  # permanent HTTP error — do not retry
        if isinstance(outcome, bytes):
            return _parse_remote_bundles(outcome)
        # Transient failure (network/timeout): record it, then back off and retry.
        errors = [outcome]
        _log_failed_attempt(attempt, attempts, outcome, backoff)

    return BundlesDoc(None, errors)

load_local_bundles(bundles_path)

Load and parse a local template-bundles file into a :class:BundlesDoc.

This is the local-file counterpart to :func:fetch_remote_bundles and part of this module's cross-module surface — :mod:rhiza_hooks._bundles_validate calls it to load a document before validating it.

Source code in rhiza_hooks/_bundles_fetch.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def load_local_bundles(bundles_path: Path) -> BundlesDoc:
    """Load and parse a local template-bundles file into a :class:`BundlesDoc`.

    This is the local-file counterpart to :func:`fetch_remote_bundles` and part
    of this module's cross-module surface — :mod:`rhiza_hooks._bundles_validate`
    calls it to load a document before validating it.
    """
    result = load_yaml_mapping(bundles_path)
    if not isinstance(result, YamlFailure):
        return BundlesDoc(result, [])

    messages = {
        YamlError.NOT_FOUND: f"Template bundles file not found: {bundles_path}",
        YamlError.INVALID: f"Invalid YAML: {result.detail}",
        YamlError.EMPTY: "Template bundles file is empty",
        YamlError.NOT_MAPPING: "Template bundles file must be a dictionary",
    }
    return BundlesDoc(None, [messages[result.kind]])