Skip to content

discovery

napt.discovery.base

Discovery strategy protocol and shared helpers.

A discovery strategy answers a single question: "what is the latest version of this app, and where can it be downloaded from?" Strategies return that answer as a RemoteVersion dataclass. They do not download files themselves; the orchestrator does.

Built-in strategies
  • api_github: queries the GitHub releases API for the latest tag.
  • api_json: extracts version and download URL from a JSON endpoint.
  • web_scrape: parses a vendor download page for both fields.

The fourth flow (url_download) is not a registered strategy. It downloads a fixed URL and extracts the version from the file itself, which is a different shape than the strategies in this module. The discovery orchestrator dispatches to that flow directly when a recipe uses strategy: url_download.

Design Philosophy
  • Strategies are typing.Protocol types. Implementations are matched structurally; no inheritance is required.
  • Strategies are pure functions of configuration. They have no state and no I/O of files.
  • Dispatch is an explicit name-to-class table in napt.discovery.registry.
  • resolve_installer turns a RemoteVersion into a StrategyResult by reusing the previous download or fetching a new one. Strategies don't call it themselves; the orchestrator does.

RemoteVersion dataclass

Version and download URL discovered from a remote source.

Returned by every DiscoveryStrategy implementation. The orchestrator passes this to resolve_installer to decide whether the file needs to be downloaded.

Attributes:

Name Type Description
version str

Raw version string extracted from the remote source (for example, "140.0.7339.128"). The trigger for a download; an MSI or MSIX installer's own version is what gets recorded.
download_url str

URL the installer can be fetched from.
source str

Name of the strategy that produced this result, used for logging and result reporting (for example, "api_github").
Source code in napt/discovery/base.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass(frozen=True)
class RemoteVersion:
    """Version and download URL discovered from a remote source.

    Returned by every [DiscoveryStrategy][napt.discovery.base.DiscoveryStrategy]
    implementation. The orchestrator passes this to
    [resolve_installer][napt.discovery.resolve.resolve_installer] to decide
    whether the file needs to be downloaded.

    Attributes:
        version: Raw version string extracted from the remote source
            (for example, ``"140.0.7339.128"``). The trigger for a
            download; an MSI or MSIX installer's own version is what gets
            recorded.
        download_url: URL the installer can be fetched from.
        source: Name of the strategy that produced this result, used
            for logging and result reporting (for example, ``"api_github"``).
    """

    version: str
    download_url: str
    source: str

StrategyResult dataclass

Resolved discovery result, ready to be recorded in deployment state.

Returned by resolve_installer for every flow. Captures everything the orchestrator needs to record the pending release and build a public DiscoverResult.

Attributes:

Name Type Description
version str

Version string for the resolved file.
version_source str

Where the version came from: "msi" or "msix" when read from the installer, otherwise the name of the strategy that reported it (for example, "api_github").
file_path Path

Path to the resolved installer on disk. This is either a freshly downloaded file or one an earlier run downloaded.
sha256 str

SHA-256 hex digest of the resolved file.
download_url str

URL the file came from, recorded with the pending release.
cached bool

True when a file from an earlier run was reused; False when it was downloaded.
Source code in napt/discovery/base.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@dataclass(frozen=True)
class StrategyResult:
    """Resolved discovery result, ready to be recorded in deployment state.

    Returned by [resolve_installer][napt.discovery.resolve.resolve_installer]
    for every flow. Captures everything the orchestrator needs to record
    the pending release and build a public
    [DiscoverResult][napt.results.DiscoverResult].

    Attributes:
        version: Version string for the resolved file.
        version_source: Where the version came from: ``"msi"`` or
            ``"msix"`` when read from the installer, otherwise the name of
            the strategy that reported it (for example, ``"api_github"``).
        file_path: Path to the resolved installer on disk. This is either
            a freshly downloaded file or one an earlier run downloaded.
        sha256: SHA-256 hex digest of the resolved file.
        download_url: URL the file came from, recorded with the pending
            release.
        cached: True when a file from an earlier run was reused; False
            when it was downloaded.
    """

    version: str
    version_source: str
    file_path: Path
    sha256: str
    download_url: str
    cached: bool

DiscoveryStrategy

Bases: Protocol

Protocol for version discovery strategies.

A strategy queries a remote source (API, web page, etc.) and returns the latest version plus its download URL. Strategies do not download files or write to disk. Those concerns belong to the orchestrator.

Implementations need only a discover and a validate_config method with the signatures below.

Source code in napt/discovery/base.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class DiscoveryStrategy(Protocol):
    """Protocol for version discovery strategies.

    A strategy queries a remote source (API, web page, etc.) and returns
    the latest version plus its download URL. Strategies do not download
    files or write to disk. Those concerns belong to the orchestrator.

    Implementations need only a ``discover`` and a ``validate_config``
    method with the signatures below.
    """

    def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
        """Discovers the latest version and its download URL.

        Args:
            app_config: Merged recipe configuration dict.

        Returns:
            Latest version, the URL it can be downloaded from, and the
            strategy's own name as the source identifier.

        Raises:
            ConfigError: On missing or invalid required configuration.
            NetworkError: On HTTP failures or version-extraction errors.

        """
        ...

    def validate_config(self, app_config: dict[str, Any]) -> list[str]:
        """Validates strategy-specific configuration fields without network calls.

        Implementations should check field presence, types, and format only.

        Args:
            app_config: Merged recipe configuration dict.

        Returns:
            Human-readable error messages. Empty when configuration is valid.

        """
        ...

discover 

discover(app_config: dict[str, Any]) -> RemoteVersion

Discovers the latest version and its download URL.

Parameters:

Name Type Description Default
app_config dict[str, Any]

Merged recipe configuration dict.

 required

Returns:

Type Description
RemoteVersion

Latest version, the URL it can be downloaded from, and the
RemoteVersion

strategy's own name as the source identifier.

Raises:

Type Description
ConfigError

On missing or invalid required configuration.
NetworkError

On HTTP failures or version-extraction errors.
Source code in napt/discovery/base.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
    """Discovers the latest version and its download URL.

    Args:
        app_config: Merged recipe configuration dict.

    Returns:
        Latest version, the URL it can be downloaded from, and the
        strategy's own name as the source identifier.

    Raises:
        ConfigError: On missing or invalid required configuration.
        NetworkError: On HTTP failures or version-extraction errors.

    """
    ...

validate_config 

validate_config(app_config: dict[str, Any]) -> list[str]

Validates strategy-specific configuration fields without network calls.

Implementations should check field presence, types, and format only.

Parameters:

Name Type Description Default
app_config dict[str, Any]

Merged recipe configuration dict.

 required

Returns:

Type Description
list[str]

Human-readable error messages. Empty when configuration is valid.
Source code in napt/discovery/base.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
def validate_config(self, app_config: dict[str, Any]) -> list[str]:
    """Validates strategy-specific configuration fields without network calls.

    Implementations should check field presence, types, and format only.

    Args:
        app_config: Merged recipe configuration dict.

    Returns:
        Human-readable error messages. Empty when configuration is valid.

    """
    ...

require_usable_version 

require_usable_version(version: str) -> None

Rejects a version that cannot name a folder or that a device would misread.

The version names the download folder (downloads/<id>/<version>) and later the build folder, so a value with a path separator or .. is refused before any folder is created from it.

It is also what the detection and requirements scripts compare on a device, and they take each part's leading digits with a part that has none counting as 0. A version whose first part has no digits (v2.0, latest) therefore reads as version 0 there: every device would report it installed and no device would ever upgrade to it. Such a version is refused so the recipe gets fixed instead.

Parameters:

Name Type Description Default
version str

Version read from the installer, or reported by a strategy for an installer that carries no version of its own.

 required

Raises:

Type Description
ConfigError

If the version is not a plain folder name, or does not start with a digit.
Source code in napt/discovery/base.py 
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
def require_usable_version(version: str) -> None:
    """Rejects a version that cannot name a folder or that a device would misread.

    The version names the download folder (``downloads/<id>/<version>``)
    and later the build folder, so a value with a path separator or ``..``
    is refused before any folder is created from it.

    It is also what the detection and requirements scripts compare on a
    device, and they take each part's leading digits with a part that has
    none counting as 0. A version whose first part has no digits (``v2.0``,
    ``latest``) therefore reads as version 0 there: every device would
    report it installed and no device would ever upgrade to it. Such a
    version is refused so the recipe gets fixed instead.

    Args:
        version: Version read from the installer, or reported by a
            strategy for an installer that carries no version of its own.

    Raises:
        ConfigError: If the version is not a plain folder name, or does
            not start with a digit.
    """
    if not is_safe_path_component(version):
        raise ConfigError(
            f"Discovered version {version!a} cannot be used as a folder name. "
            "Versions may contain only letters, digits, '.', '-', '_', and '+'. "
            "Check the recipe's version pattern, or the installer's metadata."
        )
    if not version[0].isdigit():
        # The part from the first digit onward is what a pattern should keep.
        digits = re.search(r"\d.*", version)
        hint = (
            f"Tighten the recipe's version_pattern so it captures only "
            f"{digits.group()!a}."
            if digits
            else "Check the recipe's version_path or version_pattern."
        )
        raise ConfigError(
            f"Version {version!a} does not start with a number. Devices compare "
            "versions numerically and would read it as 0, which blocks "
            f"upgrades. {hint}"
        )

napt.discovery.registry

Explicit registry of discovery strategies.

Maps each discovery.strategy recipe value to the class implementing it. Dispatch is a plain table lookup: nothing registers itself at import time, and this module's imports are what pull in the strategy implementations.

Adding a strategy
  1. Implement DiscoveryStrategy in a new module under napt/discovery/.
  2. Import the class here and add it to the table.
Note

url_download is intentionally absent from the table. It downloads the file before it can determine the version, which does not fit the version-first contract, so the discovery orchestrator dispatches to run_url_download directly.

get_strategy 

get_strategy(name: str) -> DiscoveryStrategy

Returns a discovery strategy instance by name.

Strategies are stateless, so a fresh instance is created on every call.

Parameters:

Name Type Description Default
name str

Strategy name as written in recipe YAML under discovery.strategy. Case-sensitive.

 required

Returns:

Type Description
DiscoveryStrategy

New instance of the requested strategy.

Raises:

Type Description
ConfigError

If the name is not in the registry. The message lists the available strategies for troubleshooting.
Source code in napt/discovery/registry.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_strategy(name: str) -> DiscoveryStrategy:
    """Returns a discovery strategy instance by name.

    Strategies are stateless, so a fresh instance is created on every
    call.

    Args:
        name: Strategy name as written in recipe YAML under
            ``discovery.strategy``. Case-sensitive.

    Returns:
        New instance of the requested strategy.

    Raises:
        ConfigError: If the name is not in the registry. The message
            lists the available strategies for troubleshooting.

    """
    strategy_class = _STRATEGIES.get(name)
    if strategy_class is None:
        available = ", ".join(sorted([*_STRATEGIES, "url_download"]))
        raise ConfigError(
            f"Unknown discovery strategy: {name!r}. Available: {available}"
        )
    return strategy_class()

napt.discovery.resolve

Turns what a strategy found into an installer on disk.

Every discovery flow ends here. A strategy produces a download URL and, for the version-first strategies, the version it believes that URL serves. This module downloads the file, reads the version out of it when the installer can report one (MSI ProductVersion, MSIX Identity.Version), and files it under downloads/<id>/<version>/. The installer's own version is the one recorded, because it is what the detection and requirements scripts compare against on a device; the version a page or API reported is only the trigger that caused the download.

Skipping Downloads

Each app has one sidecar file, downloads/<id>/.download.json, recording what was last resolved: the trigger (the URL and, when the strategy supplied one, the version it reported), the server's ETag and Last-Modified, and the installer's version, name, and hash. The next run compares its trigger against that record:

  • A version-first strategy that reports the same version as last time reuses the installer without a request, provided that version and the installer's own agree as a device would compare them (4.41.106 and 4.41.106.0 do). When they disagree, the report was already wrong about this file once (a page updated before the file behind it, or a pattern capturing the wrong value), so it cannot vouch that nothing changed: the run warns and asks the server instead, as url_download does.
  • url_download has no version to compare, so it sends the ETag / Last-Modified back as a conditional request and reuses the installer when the server answers HTTP 304.

The sidecar is a hint, not a record. Anything wrong with it means one full download, never an error, and it is written only after the installer is in place.

resolve_installer 

resolve_installer(
    url: str,
    app_dir: Path,
    *,
    source: str,
    discovered_version: str | None = None
) -> StrategyResult

Resolves a strategy's finding to an installer on disk.

Reuses the app's last download when the sidecar shows the trigger is unchanged (see the module description); otherwise downloads the file, reads its version, and files it under that version.

Parameters:

Name Type Description Default
url str

URL the installer is served from.

 required
app_dir Path

The app's download directory (downloads/<id>).

 required
source str

Name of the strategy, recorded as the version source when the installer cannot report its own version.

 required
discovered_version str | None

Version the strategy reported for url, or None for url_download, which learns the version from the file.

 None

Returns:

Type Description
StrategyResult

Resolved version, its source, file path, and SHA-256 hash. The
StrategyResult

cached field is True when the previous download was reused.

Raises:

Type Description
ConfigError

If the file's version cannot be used as a folder name, or if url_download fetched a file that cannot report a version.
NetworkError

On download or version-extraction failures.
Source code in napt/discovery/resolve.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def resolve_installer(
    url: str,
    app_dir: Path,
    *,
    source: str,
    discovered_version: str | None = None,
) -> StrategyResult:
    """Resolves a strategy's finding to an installer on disk.

    Reuses the app's last download when the sidecar shows the trigger is
    unchanged (see the module description); otherwise downloads the file,
    reads its version, and files it under that version.

    Args:
        url: URL the installer is served from.
        app_dir: The app's download directory (``downloads/<id>``).
        source: Name of the strategy, recorded as the version source when
            the installer cannot report its own version.
        discovered_version: Version the strategy reported for ``url``, or
            None for ``url_download``, which learns the version from the
            file.

    Returns:
        Resolved version, its source, file path, and SHA-256 hash. The
        ``cached`` field is True when the previous download was reused.

    Raises:
        ConfigError: If the file's version cannot be used as a folder
            name, or if ``url_download`` fetched a file that cannot report
            a version.
        NetworkError: On download or version-extraction failures.

    """
    logger = get_global_logger()
    previous = _load_sidecar(app_dir)

    try:
        if discovered_version is not None:
            # The reported version is the whole trigger. The URL is not part
            # of it, because some vendor pages embed a changing token in
            # every download link.
            if previous is None or previous.discovered_version != discovered_version:
                return _download_and_file(url, app_dir, source, discovered_version)
            if compare_versions(discovered_version, previous.version) == 0:
                logger.info(
                    "DISCOVERY",
                    f"Version {discovered_version} already downloaded, "
                    f"using {previous.file_path}",
                )
                return _reuse(previous, url, source)
            # The reported version was already wrong about this file once,
            # so it cannot vouch that nothing changed. Ask the server.
            _warn_mismatch(source, discovered_version, previous.version)
            return _refresh(previous, url, app_dir, source, discovered_version)

        if previous is None or previous.url != url:
            return _download_and_file(url, app_dir, source, None)
        return _refresh(previous, url, app_dir, source, None)
    except (NetworkError, ConfigError):
        raise
    except Exception as err:
        raise NetworkError(f"Failed to download {url}: {err}") from err

napt.discovery.url_download

url_download discovery flow.

This module is intentionally not a DiscoveryStrategy. The strategies in napt.discovery.base produce a RemoteVersion from configuration alone (version-first). url_download cannot do that: it has no remote endpoint to query for the version, so it must download the installer and read the version from the file itself. The discovery orchestrator special-cases strategy: url_download and dispatches to run_url_download directly.

The download, the version read, and the reuse of an unchanged file all happen in napt.discovery.resolve, shared with the version-first strategies. With no version to compare, this flow relies on the server's ETag / Last-Modified to learn whether the file changed.

Supported File Types
  • .msi: version is read from the MSI ProductVersion property.
  • .msix: version is read from the package's Identity element.
  • Other extensions raise ConfigError. For those installers, use a version-first strategy.
Recipe Example 
discovery:
  strategy: url_download
  url: "https://vendor.example.com/installer.msi"

run_url_download 

run_url_download(
    app_config: dict[str, Any], output_dir: Path
) -> StrategyResult

Downloads a fixed URL and reads the version from the resulting file.

Parameters:

Name Type Description Default
app_config dict[str, Any]

Merged recipe configuration dict containing discovery.url and id.

 required
output_dir Path

Base directory to download into. The file lands in output_dir / app_id / version.

 required

Returns:

Type Description
StrategyResult

Resolved version, file path, and SHA-256 hash. The cached
StrategyResult

field is True when HTTP 304 was used to reuse the previously
StrategyResult

downloaded file.

Raises:

Type Description
ConfigError

If discovery.url is missing, or if the downloaded file is not an MSI or MSIX.
NetworkError

On download or version-extraction failures.
Source code in napt/discovery/url_download.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def run_url_download(
    app_config: dict[str, Any],
    output_dir: Path,
) -> StrategyResult:
    """Downloads a fixed URL and reads the version from the resulting file.

    Args:
        app_config: Merged recipe configuration dict containing
            ``discovery.url`` and ``id``.
        output_dir: Base directory to download into. The file lands
            in ``output_dir / app_id / version``.

    Returns:
        Resolved version, file path, and SHA-256 hash. The ``cached``
        field is True when HTTP 304 was used to reuse the previously
        downloaded file.

    Raises:
        ConfigError: If ``discovery.url`` is missing, or if the
            downloaded file is not an MSI or MSIX.
        NetworkError: On download or version-extraction failures.

    """
    logger = get_global_logger()
    url = app_config.get("discovery", {}).get("url")
    if not url:
        raise ConfigError("url_download strategy requires 'discovery.url' in config")

    logger.verbose("DISCOVERY", "Strategy: url_download (file-first)")
    logger.verbose("DISCOVERY", f"Source URL: {url}")

    return resolve_installer(url, output_dir / app_config["id"], source="url_download")

validate_url_download_config 

validate_url_download_config(app_config: dict[str, Any]) -> list[str]

Validates url_download configuration fields.

Called by napt.validation.validate_config to compose the url_download field rules into the overall recipe validation.

Parameters:

Name Type Description Default
app_config dict[str, Any]

Merged recipe configuration dict.

 required

Returns:

Type Description
list[str]

Human-readable error messages. Empty when configuration is valid.
Source code in napt/discovery/url_download.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def validate_url_download_config(app_config: dict[str, Any]) -> list[str]:
    """Validates url_download configuration fields.

    Called by [napt.validation.validate_config][] to compose the
    url_download field rules into the overall recipe validation.

    Args:
        app_config: Merged recipe configuration dict.

    Returns:
        Human-readable error messages. Empty when configuration is valid.

    """
    errors: list[str] = []
    source = app_config.get("discovery", {})

    if "url" not in source:
        errors.append("Missing required field: discovery.url")
    elif not isinstance(source["url"], str):
        errors.append("discovery.url must be a string")
    elif not source["url"].strip():
        errors.append("discovery.url cannot be empty")

    return errors

napt.discovery.web_scrape

Web scraping discovery strategy.

Fetches a vendor download page, locates a download link, and extracts the version from that link's URL. Use this when a vendor has neither a JSON API nor a GitHub releases feed.

Recipe Example (CSS selector — recommended): 

discovery:
  strategy: web_scrape
  page_url: "https://www.7-zip.org/download.html"
  link_selector: 'a[href$="-x64.msi"]'
  version_pattern: "7z(\\d{2})(\\d{2})-x64"
  version_format: "{0}.{1}"     # transforms ("25", "01") -> "25.01"

Recipe Example (regex fallback): 

discovery:
  strategy: web_scrape
  page_url: "https://vendor.example.com/downloads"
  link_pattern: 'href="(/files/app-v[0-9.]+-x64\\.msi)"'
  version_pattern: "app-v([0-9.]+)-x64"

Configuration Fields
  • page_url (required): URL of the page to scrape.
  • link_selector (optional): CSS selector identifying the download link's <a> element. Recommended over regex.
  • link_pattern (optional): Regex with one capture group around the link URL. Used when a CSS selector cannot pin the link down. Exactly one of link_selector / link_pattern is required.
  • version_pattern (required): Regex applied to the discovered link URL to extract the version. Capture groups are pulled out and combined with version_format.
  • version_format (optional, default "{0}"): Python format string referencing capture groups by index ({0}, {1}, ...). Use this when a single version field needs to be assembled from multiple captures.
Finding a CSS Selector
  1. Open the download page in Chrome / Edge / Firefox.
  2. Right-click the download link -> Inspect.
  3. Right-click the highlighted element -> Copy -> Copy selector.
  4. Simplify the result. Common shapes:
    • a[href$=".msi"] (links ending in .msi)
    • a[href*="x64"] (links containing "x64")
    • a.download (links with class="download")
Note

The selector / pattern is expected to match exactly one link; the first match is used. Relative URLs in the page are resolved against page_url. CSS selector support requires BeautifulSoup4; the regex fallback does not.

WebScrapeStrategy

Discovery strategy for scraping vendor download pages.

Source code in napt/discovery/web_scrape.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
class WebScrapeStrategy:
    """Discovery strategy for scraping vendor download pages."""

    def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
        r"""Discovers version and download URL by scraping a vendor page.

        Fetches ``discovery.page_url``, locates a download link with
        either ``link_selector`` (CSS) or ``link_pattern`` (regex),
        and extracts the version from the matched link using
        ``version_pattern``.

        Args:
            app_config: Merged recipe configuration dict containing
                ``discovery.page_url``, exactly one of
                ``discovery.link_selector`` or ``discovery.link_pattern``,
                and ``discovery.version_pattern``.

        Returns:
            Discovered version, the matched link's URL, and
            ``"web_scrape"`` as the source identifier.

        Raises:
            ConfigError: On missing required configuration or when
                a selector / pattern matches nothing.
            NetworkError: On page fetch failure.

        """
        from napt.logging import get_global_logger

        logger = get_global_logger()
        # Validate configuration
        source = app_config.get("discovery", {})
        page_url = source.get("page_url")
        if not page_url:
            raise ConfigError(
                "web_scrape strategy requires 'discovery.page_url' in config"
            )

        link_selector = source.get("link_selector")
        link_pattern = source.get("link_pattern")

        if not link_selector and not link_pattern:
            raise ConfigError(
                "web_scrape strategy requires either 'discovery.link_selector' or "
                "'discovery.link_pattern' in config"
            )

        version_pattern = source.get("version_pattern")
        if not version_pattern:
            raise ConfigError(
                "web_scrape strategy requires 'discovery.version_pattern' in config"
            )

        version_format = source.get("version_format", _DEFAULT_VERSION_FORMAT)

        logger.verbose("DISCOVERY", "Strategy: web_scrape (version-first)")
        logger.verbose("DISCOVERY", f"Page URL: {page_url}")
        if link_selector:
            logger.verbose("DISCOVERY", f"Link selector (CSS): {link_selector}")
        if link_pattern:
            logger.verbose("DISCOVERY", f"Link pattern (regex): {link_pattern}")
        logger.verbose("DISCOVERY", f"Version pattern: {version_pattern}")

        # Download the HTML page
        logger.verbose("DISCOVERY", f"Fetching page: {page_url}")
        try:
            with make_session() as session:
                response = session.get(page_url, timeout=30)
        except requests.exceptions.RequestException as err:
            raise NetworkError(f"Failed to fetch page: {err}") from err

        if not response.ok:
            raise NetworkError(
                f"Failed to fetch page: {response.status_code} {response.reason}"
            )

        html_content = response.text
        logger.verbose("DISCOVERY", f"Page fetched ({len(html_content)} bytes)")

        # Find download link using CSS selector or regex
        download_url = None

        if link_selector:
            # Use CSS selector with BeautifulSoup4
            soup = BeautifulSoup(html_content, "html.parser")
            element = soup.select_one(link_selector)

            if not element:
                raise ConfigError(
                    f"CSS selector {link_selector!r} did not match any elements on page"
                )

            # Get href attribute
            href = element.get("href")
            if not isinstance(href, str) or not href:
                raise ConfigError(
                    f"Element matched by {link_selector!r} has no href attribute"
                )

            logger.verbose("DISCOVERY", f"Found link via CSS: {href}")

            # Build absolute URL
            download_url = urljoin(page_url, href)

        elif link_pattern:
            # Use regex fallback
            try:
                pattern = re.compile(link_pattern)
                match = pattern.search(html_content)

                if not match:
                    raise ConfigError(
                        f"Regex pattern {link_pattern!r} did not match anything on page"
                    )

                # Get first capture group or full match
                if pattern.groups > 0:
                    href = match.group(1)
                else:
                    href = match.group(0)

                logger.verbose("DISCOVERY", f"Found link via regex: {href}")

                # Build absolute URL
                download_url = urljoin(page_url, href)

            except re.error as err:
                raise ConfigError(
                    f"Invalid link_pattern regex: {link_pattern!r}"
                ) from err

        else:
            raise ConfigError(
                "web_scrape strategy requires either 'discovery.link_selector' or "
                "'discovery.link_pattern' in config"
            )

        logger.verbose("DISCOVERY", f"Download URL: {download_url}")

        # Extract version from the download URL
        try:
            version_regex = re.compile(version_pattern)
            match = version_regex.search(download_url)

            if not match:
                raise ConfigError(
                    f"Version pattern {version_pattern!r} did not match "
                    f"URL {download_url!r}"
                )

            # Get captured groups
            groups = match.groups()

            if not groups:
                # No capture groups, use full match
                version_str = match.group(0)
            else:
                # Format using captured groups
                try:
                    version_str = version_format.format(*groups)
                except (IndexError, KeyError) as err:
                    raise ConfigError(
                        f"version_format {version_format!r} failed with "
                        f"groups {groups}: {err}"
                    ) from err

        except re.error as err:
            raise ConfigError(
                f"Invalid version_pattern regex: {version_pattern!r}"
            ) from err

        logger.verbose("DISCOVERY", f"Extracted version: {version_str}")

        return RemoteVersion(
            version=version_str,
            download_url=download_url,
            source="web_scrape",
        )

    def validate_config(self, app_config: dict[str, Any]) -> list[str]:
        """Validate web_scrape strategy configuration.

        Checks for required fields and correct types without making network calls.

        Args:
            app_config: The app configuration from the recipe.

        Returns:
            List of error messages (empty if valid).

        """
        errors = []
        source = app_config.get("discovery", {})

        # Check page_url
        if "page_url" not in source:
            errors.append("Missing required field: discovery.page_url")
        elif not isinstance(source["page_url"], str):
            errors.append("discovery.page_url must be a string")
        elif not source["page_url"].strip():
            errors.append("discovery.page_url cannot be empty")

        # Check that at least one link finding method is provided
        link_selector = source.get("link_selector")
        link_pattern = source.get("link_pattern")

        if not link_selector and not link_pattern:
            errors.append(
                "Missing required field: must provide either "
                "discovery.link_selector or discovery.link_pattern"
            )

        # Validate link_selector if provided
        if link_selector:
            if not isinstance(link_selector, str):
                errors.append("discovery.link_selector must be a string")
            elif not link_selector.strip():
                errors.append("discovery.link_selector cannot be empty")
            else:
                # Try to validate CSS selector syntax
                try:
                    # Test if selector is parseable
                    soup = BeautifulSoup("<html></html>", "html.parser")
                    soup.select_one(link_selector)  # Will raise if invalid
                except Exception as err:
                    errors.append(f"Invalid CSS selector: {err}")

        # Validate link_pattern if provided
        if link_pattern:
            if not isinstance(link_pattern, str):
                errors.append("discovery.link_pattern must be a string")
            elif not link_pattern.strip():
                errors.append("discovery.link_pattern cannot be empty")
            else:
                # Validate regex compiles
                try:
                    re.compile(link_pattern)
                except re.error as err:
                    errors.append(f"Invalid link_pattern regex: {err}")

        # Check version_pattern
        if "version_pattern" not in source:
            errors.append("Missing required field: discovery.version_pattern")
        elif not isinstance(source["version_pattern"], str):
            errors.append("discovery.version_pattern must be a string")
        elif not source["version_pattern"].strip():
            errors.append("discovery.version_pattern cannot be empty")
        else:
            # Validate regex compiles
            try:
                re.compile(source["version_pattern"])
            except re.error as err:
                errors.append(f"Invalid version_pattern regex: {err}")

        # Validate version_format if provided
        if "version_format" in source:
            if not isinstance(source["version_format"], str):
                errors.append("discovery.version_format must be a string")
            elif not source["version_format"].strip():
                errors.append("discovery.version_format cannot be empty")

        return errors

discover 

discover(app_config: dict[str, Any]) -> RemoteVersion

Discovers version and download URL by scraping a vendor page.

Fetches discovery.page_url, locates a download link with either link_selector (CSS) or link_pattern (regex), and extracts the version from the matched link using version_pattern.

Parameters:

Name Type Description Default
app_config dict[str, Any]

Merged recipe configuration dict containing discovery.page_url, exactly one of discovery.link_selector or discovery.link_pattern, and discovery.version_pattern.

 required

Returns:

Type Description
RemoteVersion

Discovered version, the matched link's URL, and
RemoteVersion

"web_scrape" as the source identifier.

Raises:

Type Description
ConfigError

On missing required configuration or when a selector / pattern matches nothing.
NetworkError

On page fetch failure.
Source code in napt/discovery/web_scrape.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
    r"""Discovers version and download URL by scraping a vendor page.

    Fetches ``discovery.page_url``, locates a download link with
    either ``link_selector`` (CSS) or ``link_pattern`` (regex),
    and extracts the version from the matched link using
    ``version_pattern``.

    Args:
        app_config: Merged recipe configuration dict containing
            ``discovery.page_url``, exactly one of
            ``discovery.link_selector`` or ``discovery.link_pattern``,
            and ``discovery.version_pattern``.

    Returns:
        Discovered version, the matched link's URL, and
        ``"web_scrape"`` as the source identifier.

    Raises:
        ConfigError: On missing required configuration or when
            a selector / pattern matches nothing.
        NetworkError: On page fetch failure.

    """
    from napt.logging import get_global_logger

    logger = get_global_logger()
    # Validate configuration
    source = app_config.get("discovery", {})
    page_url = source.get("page_url")
    if not page_url:
        raise ConfigError(
            "web_scrape strategy requires 'discovery.page_url' in config"
        )

    link_selector = source.get("link_selector")
    link_pattern = source.get("link_pattern")

    if not link_selector and not link_pattern:
        raise ConfigError(
            "web_scrape strategy requires either 'discovery.link_selector' or "
            "'discovery.link_pattern' in config"
        )

    version_pattern = source.get("version_pattern")
    if not version_pattern:
        raise ConfigError(
            "web_scrape strategy requires 'discovery.version_pattern' in config"
        )

    version_format = source.get("version_format", _DEFAULT_VERSION_FORMAT)

    logger.verbose("DISCOVERY", "Strategy: web_scrape (version-first)")
    logger.verbose("DISCOVERY", f"Page URL: {page_url}")
    if link_selector:
        logger.verbose("DISCOVERY", f"Link selector (CSS): {link_selector}")
    if link_pattern:
        logger.verbose("DISCOVERY", f"Link pattern (regex): {link_pattern}")
    logger.verbose("DISCOVERY", f"Version pattern: {version_pattern}")

    # Download the HTML page
    logger.verbose("DISCOVERY", f"Fetching page: {page_url}")
    try:
        with make_session() as session:
            response = session.get(page_url, timeout=30)
    except requests.exceptions.RequestException as err:
        raise NetworkError(f"Failed to fetch page: {err}") from err

    if not response.ok:
        raise NetworkError(
            f"Failed to fetch page: {response.status_code} {response.reason}"
        )

    html_content = response.text
    logger.verbose("DISCOVERY", f"Page fetched ({len(html_content)} bytes)")

    # Find download link using CSS selector or regex
    download_url = None

    if link_selector:
        # Use CSS selector with BeautifulSoup4
        soup = BeautifulSoup(html_content, "html.parser")
        element = soup.select_one(link_selector)

        if not element:
            raise ConfigError(
                f"CSS selector {link_selector!r} did not match any elements on page"
            )

        # Get href attribute
        href = element.get("href")
        if not isinstance(href, str) or not href:
            raise ConfigError(
                f"Element matched by {link_selector!r} has no href attribute"
            )

        logger.verbose("DISCOVERY", f"Found link via CSS: {href}")

        # Build absolute URL
        download_url = urljoin(page_url, href)

    elif link_pattern:
        # Use regex fallback
        try:
            pattern = re.compile(link_pattern)
            match = pattern.search(html_content)

            if not match:
                raise ConfigError(
                    f"Regex pattern {link_pattern!r} did not match anything on page"
                )

            # Get first capture group or full match
            if pattern.groups > 0:
                href = match.group(1)
            else:
                href = match.group(0)

            logger.verbose("DISCOVERY", f"Found link via regex: {href}")

            # Build absolute URL
            download_url = urljoin(page_url, href)

        except re.error as err:
            raise ConfigError(
                f"Invalid link_pattern regex: {link_pattern!r}"
            ) from err

    else:
        raise ConfigError(
            "web_scrape strategy requires either 'discovery.link_selector' or "
            "'discovery.link_pattern' in config"
        )

    logger.verbose("DISCOVERY", f"Download URL: {download_url}")

    # Extract version from the download URL
    try:
        version_regex = re.compile(version_pattern)
        match = version_regex.search(download_url)

        if not match:
            raise ConfigError(
                f"Version pattern {version_pattern!r} did not match "
                f"URL {download_url!r}"
            )

        # Get captured groups
        groups = match.groups()

        if not groups:
            # No capture groups, use full match
            version_str = match.group(0)
        else:
            # Format using captured groups
            try:
                version_str = version_format.format(*groups)
            except (IndexError, KeyError) as err:
                raise ConfigError(
                    f"version_format {version_format!r} failed with "
                    f"groups {groups}: {err}"
                ) from err

    except re.error as err:
        raise ConfigError(
            f"Invalid version_pattern regex: {version_pattern!r}"
        ) from err

    logger.verbose("DISCOVERY", f"Extracted version: {version_str}")

    return RemoteVersion(
        version=version_str,
        download_url=download_url,
        source="web_scrape",
    )

validate_config 

validate_config(app_config: dict[str, Any]) -> list[str]

Validate web_scrape strategy configuration.

Checks for required fields and correct types without making network calls.

Parameters:

Name Type Description Default
app_config dict[str, Any]

The app configuration from the recipe.

 required

Returns:

Type Description
list[str]

List of error messages (empty if valid).
Source code in napt/discovery/web_scrape.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def validate_config(self, app_config: dict[str, Any]) -> list[str]:
    """Validate web_scrape strategy configuration.

    Checks for required fields and correct types without making network calls.

    Args:
        app_config: The app configuration from the recipe.

    Returns:
        List of error messages (empty if valid).

    """
    errors = []
    source = app_config.get("discovery", {})

    # Check page_url
    if "page_url" not in source:
        errors.append("Missing required field: discovery.page_url")
    elif not isinstance(source["page_url"], str):
        errors.append("discovery.page_url must be a string")
    elif not source["page_url"].strip():
        errors.append("discovery.page_url cannot be empty")

    # Check that at least one link finding method is provided
    link_selector = source.get("link_selector")
    link_pattern = source.get("link_pattern")

    if not link_selector and not link_pattern:
        errors.append(
            "Missing required field: must provide either "
            "discovery.link_selector or discovery.link_pattern"
        )

    # Validate link_selector if provided
    if link_selector:
        if not isinstance(link_selector, str):
            errors.append("discovery.link_selector must be a string")
        elif not link_selector.strip():
            errors.append("discovery.link_selector cannot be empty")
        else:
            # Try to validate CSS selector syntax
            try:
                # Test if selector is parseable
                soup = BeautifulSoup("<html></html>", "html.parser")
                soup.select_one(link_selector)  # Will raise if invalid
            except Exception as err:
                errors.append(f"Invalid CSS selector: {err}")

    # Validate link_pattern if provided
    if link_pattern:
        if not isinstance(link_pattern, str):
            errors.append("discovery.link_pattern must be a string")
        elif not link_pattern.strip():
            errors.append("discovery.link_pattern cannot be empty")
        else:
            # Validate regex compiles
            try:
                re.compile(link_pattern)
            except re.error as err:
                errors.append(f"Invalid link_pattern regex: {err}")

    # Check version_pattern
    if "version_pattern" not in source:
        errors.append("Missing required field: discovery.version_pattern")
    elif not isinstance(source["version_pattern"], str):
        errors.append("discovery.version_pattern must be a string")
    elif not source["version_pattern"].strip():
        errors.append("discovery.version_pattern cannot be empty")
    else:
        # Validate regex compiles
        try:
            re.compile(source["version_pattern"])
        except re.error as err:
            errors.append(f"Invalid version_pattern regex: {err}")

    # Validate version_format if provided
    if "version_format" in source:
        if not isinstance(source["version_format"], str):
            errors.append("discovery.version_format must be a string")
        elif not source["version_format"].strip():
            errors.append("discovery.version_format cannot be empty")

    return errors

napt.discovery.api_github

GitHub releases discovery strategy.

Queries the GitHub releases API for the latest tag and the download URL of a matching asset. The version comes from the release tag (parsed with a regex); the download URL comes from the first asset whose filename matches asset_pattern.

Recipe Example 
discovery:
  strategy: api_github
  repo: "git-for-windows/git"            # required, "owner/name"
  asset_pattern: "Git-.*-64-bit\\.exe$"  # required, regex on asset filename
  version_pattern: "v?([0-9.]+)"         # optional, default strips "v"
  prerelease: false                      # optional, default false
  token: "${GITHUB_TOKEN}"               # optional, supports env expansion
Configuration Fields
  • repo (required): GitHub repo as "owner/name".
  • asset_pattern (required): Regex matched against asset filename. First match wins. Case-sensitive by default; prefix with (?i) for case-insensitive matching.
  • version_pattern (optional): Regex for extracting the version from the release tag. Uses capture group 1 if present, otherwise the full match. Default: v?([0-9.]+).
  • prerelease (optional, default false): When true, includes pre-release versions; otherwise the latest release must be stable.
  • token (optional): GitHub personal access token. Raises the API rate limit from 60 to 5000 requests/hour. Supports ${ENV_VAR} expansion. Public repos do not require any special permissions.
Note

GitHub returns the most recent release first. If no asset matches, or the latest release is a pre-release while prerelease: false, discovery raises an error rather than walking back through history.

ApiGithubStrategy

Discovery strategy for GitHub releases.

Source code in napt/discovery/api_github.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
class ApiGithubStrategy:
    """Discovery strategy for GitHub releases."""

    def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
        r"""Discovers the latest GitHub release version and asset download URL.

        Queries the GitHub releases API for the latest release of the
        configured repository. Extracts the version from the release tag
        (via ``version_pattern``) and the download URL from the first
        asset matching ``asset_pattern``.

        Args:
            app_config: Merged recipe configuration dict containing
                ``discovery.repo`` and ``discovery.asset_pattern``,
                plus optional ``version_pattern``, ``prerelease``, and
                ``token`` fields.

        Returns:
            Latest version, the matched asset's download URL, and
            ``"api_github"`` as the source identifier.

        Raises:
            ConfigError: On missing or malformed required configuration,
                or when patterns do not match the release.
            NetworkError: On API failure, missing assets, or rejected
                pre-releases.

        """
        from napt.logging import get_global_logger

        logger = get_global_logger()
        # Validate configuration
        source = app_config.get("discovery", {})
        repo = source.get("repo")
        if not repo:
            raise ConfigError("api_github strategy requires 'discovery.repo' in config")

        # Validate repo format
        if "/" not in repo or repo.count("/") != 1:
            raise ConfigError(
                f"Invalid repo format: {repo!r}. Expected 'owner/repository'"
            )

        # Optional configuration
        asset_pattern = source.get("asset_pattern")
        if not asset_pattern:
            raise ConfigError(
                "api_github strategy requires 'discovery.asset_pattern' in config"
            )

        version_pattern = source.get("version_pattern", _DEFAULT_VERSION_PATTERN)
        prerelease = source.get("prerelease", _DEFAULT_PRERELEASE)
        token = source.get("token")

        # Expand environment variables in token (e.g., ${GITHUB_TOKEN})
        if token:
            if token.startswith("${") and token.endswith("}"):
                env_var = token[2:-1]
                token = os.environ.get(env_var)
                if not token:
                    logger.verbose(
                        "DISCOVERY",
                        f"Warning: Environment variable {env_var} not set",
                    )

        logger.verbose("DISCOVERY", "Strategy: api_github (version-first)")
        logger.verbose("DISCOVERY", f"Repository: {repo}")
        logger.verbose("DISCOVERY", f"Version pattern: {version_pattern}")
        if asset_pattern:
            logger.verbose("DISCOVERY", f"Asset pattern: {asset_pattern}")
        if prerelease:
            logger.verbose("DISCOVERY", "Including pre-releases")

        # Fetch latest release from GitHub API
        api_url = f"https://api.github.com/repos/{repo}/releases/latest"
        headers = {
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }

        # Add authentication if token provided
        if token:
            headers["Authorization"] = f"token {token}"
            logger.verbose("DISCOVERY", "Using authenticated API request")

        logger.verbose("DISCOVERY", f"Fetching release from: {api_url}")

        try:
            with make_session() as session:
                response = session.get(api_url, headers=headers, timeout=30)
        except requests.exceptions.RequestException as err:
            raise NetworkError(f"Failed to fetch GitHub release: {err}") from err

        if response.status_code == 404:
            raise NetworkError(f"Repository {repo!r} not found or has no releases")
        elif response.status_code == 403:
            raise NetworkError(
                f"GitHub API rate limit exceeded. Consider using a token. "
                f"Status: {response.status_code}"
            )
        elif not response.ok:
            raise NetworkError(
                f"GitHub API request failed: {response.status_code} "
                f"{response.reason}"
            )

        release_data = response.json()

        # Check if this is a prerelease and we don't want those
        if release_data.get("prerelease", False) and not prerelease:
            raise NetworkError(
                f"Latest release is a pre-release and prerelease=false. "
                f"Tag: {release_data.get('tag_name')}"
            )

        # Extract version from tag name
        tag_name = release_data.get("tag_name", "")
        if not tag_name:
            raise NetworkError("Release has no tag_name field")

        logger.verbose("DISCOVERY", f"Release tag: {tag_name}")

        try:
            pattern = re.compile(version_pattern)
            match = pattern.search(tag_name)
            if not match:
                raise ConfigError(
                    f"Version pattern {version_pattern!r} did not match "
                    f"tag {tag_name!r}"
                )

            # Capture group 1 if present, else the full match
            if pattern.groups > 0:
                version_str = match.group(1)
            else:
                version_str = match.group(0)

        except re.error as err:
            raise ConfigError(
                f"Invalid version_pattern regex: {version_pattern!r}"
            ) from err
        except (ValueError, IndexError) as err:
            raise ConfigError(
                f"Failed to extract version from tag {tag_name!r} "
                f"using pattern {version_pattern!r}: {err}"
            ) from err

        logger.verbose("DISCOVERY", f"Extracted version: {version_str}")

        # Find matching asset
        assets = release_data.get("assets", [])
        if not assets:
            raise NetworkError(
                f"Release {tag_name} has no assets. "
                f"Check if assets were uploaded to the release."
            )

        logger.verbose("DISCOVERY", f"Release has {len(assets)} asset(s)")

        # Match asset by pattern
        matched_asset = None
        try:
            pattern = re.compile(asset_pattern)
        except re.error as err:
            raise ConfigError(
                f"Invalid asset_pattern regex: {asset_pattern!r}"
            ) from err

        for asset in assets:
            asset_name = asset.get("name", "")
            if pattern.search(asset_name):
                matched_asset = asset
                logger.verbose("DISCOVERY", f"Matched asset: {asset_name}")
                break

        if not matched_asset:
            available = [a.get("name", "(unnamed)") for a in assets]
            raise ConfigError(
                f"No assets matched pattern {asset_pattern!r}. "
                f"Available assets: {', '.join(available)}"
            )

        # Get download URL
        download_url = matched_asset.get("browser_download_url")
        if not download_url:
            raise NetworkError(f"Asset {matched_asset.get('name')} has no download URL")

        logger.verbose("DISCOVERY", f"Download URL: {download_url}")

        return RemoteVersion(
            version=version_str,
            download_url=download_url,
            source="api_github",
        )

    def validate_config(self, app_config: dict[str, Any]) -> list[str]:
        """Validate api_github strategy configuration.

        Checks for required fields and correct types without making network calls.

        Args:
            app_config: The app configuration from the recipe.

        Returns:
            List of error messages (empty if valid).

        """
        errors = []
        source = app_config.get("discovery", {})

        # Check required fields
        if "repo" not in source:
            errors.append("Missing required field: discovery.repo")
        elif not isinstance(source["repo"], str):
            errors.append("discovery.repo must be a string")
        elif not source["repo"].strip():
            errors.append("discovery.repo cannot be empty")
        else:
            # Validate repo format
            repo = source["repo"]
            if repo.count("/") != 1:
                errors.append(
                    "discovery.repo must be in format 'owner/repo' (e.g., 'git/git')"
                )

        if "asset_pattern" not in source:
            errors.append("Missing required field: discovery.asset_pattern")
        elif not isinstance(source["asset_pattern"], str):
            errors.append("discovery.asset_pattern must be a string")
        elif not source["asset_pattern"].strip():
            errors.append("discovery.asset_pattern cannot be empty")
        else:
            # Validate regex pattern syntax
            pattern = source["asset_pattern"]
            import re

            try:
                re.compile(pattern)
            except re.error as err:
                errors.append(f"Invalid asset_pattern regex: {err}")

        # Optional fields validation
        if "version_pattern" in source:
            if not isinstance(source["version_pattern"], str):
                errors.append("discovery.version_pattern must be a string")
            else:
                pattern = source["version_pattern"]
                import re

                try:
                    re.compile(pattern)
                except re.error as err:
                    errors.append(f"Invalid version_pattern regex: {err}")

        return errors

discover 

discover(app_config: dict[str, Any]) -> RemoteVersion

Discovers the latest GitHub release version and asset download URL.

Queries the GitHub releases API for the latest release of the configured repository. Extracts the version from the release tag (via version_pattern) and the download URL from the first asset matching asset_pattern.

Parameters:

Name Type Description Default
app_config dict[str, Any]

Merged recipe configuration dict containing discovery.repo and discovery.asset_pattern, plus optional version_pattern, prerelease, and token fields.

 required

Returns:

Type Description
RemoteVersion

Latest version, the matched asset's download URL, and
RemoteVersion

"api_github" as the source identifier.

Raises:

Type Description
ConfigError

On missing or malformed required configuration, or when patterns do not match the release.
NetworkError

On API failure, missing assets, or rejected pre-releases.
Source code in napt/discovery/api_github.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
    r"""Discovers the latest GitHub release version and asset download URL.

    Queries the GitHub releases API for the latest release of the
    configured repository. Extracts the version from the release tag
    (via ``version_pattern``) and the download URL from the first
    asset matching ``asset_pattern``.

    Args:
        app_config: Merged recipe configuration dict containing
            ``discovery.repo`` and ``discovery.asset_pattern``,
            plus optional ``version_pattern``, ``prerelease``, and
            ``token`` fields.

    Returns:
        Latest version, the matched asset's download URL, and
        ``"api_github"`` as the source identifier.

    Raises:
        ConfigError: On missing or malformed required configuration,
            or when patterns do not match the release.
        NetworkError: On API failure, missing assets, or rejected
            pre-releases.

    """
    from napt.logging import get_global_logger

    logger = get_global_logger()
    # Validate configuration
    source = app_config.get("discovery", {})
    repo = source.get("repo")
    if not repo:
        raise ConfigError("api_github strategy requires 'discovery.repo' in config")

    # Validate repo format
    if "/" not in repo or repo.count("/") != 1:
        raise ConfigError(
            f"Invalid repo format: {repo!r}. Expected 'owner/repository'"
        )

    # Optional configuration
    asset_pattern = source.get("asset_pattern")
    if not asset_pattern:
        raise ConfigError(
            "api_github strategy requires 'discovery.asset_pattern' in config"
        )

    version_pattern = source.get("version_pattern", _DEFAULT_VERSION_PATTERN)
    prerelease = source.get("prerelease", _DEFAULT_PRERELEASE)
    token = source.get("token")

    # Expand environment variables in token (e.g., ${GITHUB_TOKEN})
    if token:
        if token.startswith("${") and token.endswith("}"):
            env_var = token[2:-1]
            token = os.environ.get(env_var)
            if not token:
                logger.verbose(
                    "DISCOVERY",
                    f"Warning: Environment variable {env_var} not set",
                )

    logger.verbose("DISCOVERY", "Strategy: api_github (version-first)")
    logger.verbose("DISCOVERY", f"Repository: {repo}")
    logger.verbose("DISCOVERY", f"Version pattern: {version_pattern}")
    if asset_pattern:
        logger.verbose("DISCOVERY", f"Asset pattern: {asset_pattern}")
    if prerelease:
        logger.verbose("DISCOVERY", "Including pre-releases")

    # Fetch latest release from GitHub API
    api_url = f"https://api.github.com/repos/{repo}/releases/latest"
    headers = {
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }

    # Add authentication if token provided
    if token:
        headers["Authorization"] = f"token {token}"
        logger.verbose("DISCOVERY", "Using authenticated API request")

    logger.verbose("DISCOVERY", f"Fetching release from: {api_url}")

    try:
        with make_session() as session:
            response = session.get(api_url, headers=headers, timeout=30)
    except requests.exceptions.RequestException as err:
        raise NetworkError(f"Failed to fetch GitHub release: {err}") from err

    if response.status_code == 404:
        raise NetworkError(f"Repository {repo!r} not found or has no releases")
    elif response.status_code == 403:
        raise NetworkError(
            f"GitHub API rate limit exceeded. Consider using a token. "
            f"Status: {response.status_code}"
        )
    elif not response.ok:
        raise NetworkError(
            f"GitHub API request failed: {response.status_code} "
            f"{response.reason}"
        )

    release_data = response.json()

    # Check if this is a prerelease and we don't want those
    if release_data.get("prerelease", False) and not prerelease:
        raise NetworkError(
            f"Latest release is a pre-release and prerelease=false. "
            f"Tag: {release_data.get('tag_name')}"
        )

    # Extract version from tag name
    tag_name = release_data.get("tag_name", "")
    if not tag_name:
        raise NetworkError("Release has no tag_name field")

    logger.verbose("DISCOVERY", f"Release tag: {tag_name}")

    try:
        pattern = re.compile(version_pattern)
        match = pattern.search(tag_name)
        if not match:
            raise ConfigError(
                f"Version pattern {version_pattern!r} did not match "
                f"tag {tag_name!r}"
            )

        # Capture group 1 if present, else the full match
        if pattern.groups > 0:
            version_str = match.group(1)
        else:
            version_str = match.group(0)

    except re.error as err:
        raise ConfigError(
            f"Invalid version_pattern regex: {version_pattern!r}"
        ) from err
    except (ValueError, IndexError) as err:
        raise ConfigError(
            f"Failed to extract version from tag {tag_name!r} "
            f"using pattern {version_pattern!r}: {err}"
        ) from err

    logger.verbose("DISCOVERY", f"Extracted version: {version_str}")

    # Find matching asset
    assets = release_data.get("assets", [])
    if not assets:
        raise NetworkError(
            f"Release {tag_name} has no assets. "
            f"Check if assets were uploaded to the release."
        )

    logger.verbose("DISCOVERY", f"Release has {len(assets)} asset(s)")

    # Match asset by pattern
    matched_asset = None
    try:
        pattern = re.compile(asset_pattern)
    except re.error as err:
        raise ConfigError(
            f"Invalid asset_pattern regex: {asset_pattern!r}"
        ) from err

    for asset in assets:
        asset_name = asset.get("name", "")
        if pattern.search(asset_name):
            matched_asset = asset
            logger.verbose("DISCOVERY", f"Matched asset: {asset_name}")
            break

    if not matched_asset:
        available = [a.get("name", "(unnamed)") for a in assets]
        raise ConfigError(
            f"No assets matched pattern {asset_pattern!r}. "
            f"Available assets: {', '.join(available)}"
        )

    # Get download URL
    download_url = matched_asset.get("browser_download_url")
    if not download_url:
        raise NetworkError(f"Asset {matched_asset.get('name')} has no download URL")

    logger.verbose("DISCOVERY", f"Download URL: {download_url}")

    return RemoteVersion(
        version=version_str,
        download_url=download_url,
        source="api_github",
    )

validate_config 

validate_config(app_config: dict[str, Any]) -> list[str]

Validate api_github strategy configuration.

Checks for required fields and correct types without making network calls.

Parameters:

Name Type Description Default
app_config dict[str, Any]

The app configuration from the recipe.

 required

Returns:

Type Description
list[str]

List of error messages (empty if valid).
Source code in napt/discovery/api_github.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def validate_config(self, app_config: dict[str, Any]) -> list[str]:
    """Validate api_github strategy configuration.

    Checks for required fields and correct types without making network calls.

    Args:
        app_config: The app configuration from the recipe.

    Returns:
        List of error messages (empty if valid).

    """
    errors = []
    source = app_config.get("discovery", {})

    # Check required fields
    if "repo" not in source:
        errors.append("Missing required field: discovery.repo")
    elif not isinstance(source["repo"], str):
        errors.append("discovery.repo must be a string")
    elif not source["repo"].strip():
        errors.append("discovery.repo cannot be empty")
    else:
        # Validate repo format
        repo = source["repo"]
        if repo.count("/") != 1:
            errors.append(
                "discovery.repo must be in format 'owner/repo' (e.g., 'git/git')"
            )

    if "asset_pattern" not in source:
        errors.append("Missing required field: discovery.asset_pattern")
    elif not isinstance(source["asset_pattern"], str):
        errors.append("discovery.asset_pattern must be a string")
    elif not source["asset_pattern"].strip():
        errors.append("discovery.asset_pattern cannot be empty")
    else:
        # Validate regex pattern syntax
        pattern = source["asset_pattern"]
        import re

        try:
            re.compile(pattern)
        except re.error as err:
            errors.append(f"Invalid asset_pattern regex: {err}")

    # Optional fields validation
    if "version_pattern" in source:
        if not isinstance(source["version_pattern"], str):
            errors.append("discovery.version_pattern must be a string")
        else:
            pattern = source["version_pattern"]
            import re

            try:
                re.compile(pattern)
            except re.error as err:
                errors.append(f"Invalid version_pattern regex: {err}")

    return errors

napt.discovery.api_json

JSON API discovery strategy.

Queries a JSON API endpoint for the latest version and download URL. Both fields are extracted from the response using JSONPath expressions.

Recipe Example 
discovery:
  strategy: api_json
  api_url: "https://vendor.example.com/api/latest"  # required
  version_path: "version"                           # required, JSONPath
  download_url_path: "download_url"                 # required, JSONPath
  version_pattern: "v?([0-9.]+)"                    # optional, regex
  headers:                                          # optional
    Authorization: "Bearer ${API_TOKEN}"
    Accept: "application/json"

Nested response, with auth header: 

discovery:
  strategy: api_json
  api_url: "https://vendor.example.com/api/releases"
  version_path: "stable.version"
  download_url_path: "stable.platforms.windows.x64"
  headers:
    Authorization: "Bearer ${API_TOKEN}"

Configuration Fields
  • api_url (required): JSON endpoint URL.
  • version_path (required): JSONPath expression locating the version string in the response (e.g. "version", "release.version").
  • download_url_path (required): JSONPath expression locating the installer download URL in the response.
  • version_pattern (optional): Regex applied to the value found at version_path. Uses capture group 1 if present, otherwise the full match. Without it the value is used as is; add it when the API wraps the version in a prefix or suffix ("v2.0", "2.0 (stable)").
  • headers (optional): HTTP headers to send. Values support ${ENV_VAR} expansion.
Note

JSONPath uses the jsonpath-ng library. Environment-variable expansion (${VAR}) is applied to string values in headers.

ApiJsonStrategy

Discovery strategy for JSON API endpoints.

Source code in napt/discovery/api_json.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
class ApiJsonStrategy:
    """Discovery strategy for JSON API endpoints."""

    def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
        """Discovers version and download URL from a JSON API endpoint.

        Sends a GET request to the configured ``api_url`` and extracts
        the version and download URL using JSONPath expressions.

        Args:
            app_config: Merged recipe configuration dict containing
                ``discovery.api_url``, ``discovery.version_path``, and
                ``discovery.download_url_path``, plus optional
                ``version_pattern`` and ``headers`` fields.

        Returns:
            Discovered version, download URL, and ``"api_json"`` as
            the source identifier.

        Raises:
            ConfigError: On missing required configuration, when the
                JSONPath expressions do not match the response, or when
                ``version_pattern`` is invalid or does not match.
            NetworkError: On API request failure.

        """
        from napt.logging import get_global_logger

        logger = get_global_logger()
        # Validate configuration
        source = app_config.get("discovery", {})
        api_url = source.get("api_url")
        if not api_url:
            raise ConfigError(
                "api_json strategy requires 'discovery.api_url' in config"
            )

        version_path = source.get("version_path")
        if not version_path:
            raise ConfigError(
                "api_json strategy requires 'discovery.version_path' in config"
            )

        download_url_path = source.get("download_url_path")
        if not download_url_path:
            raise ConfigError(
                "api_json strategy requires 'discovery.download_url_path' in config"
            )

        # Optional configuration
        headers = source.get("headers", {})

        logger.verbose("DISCOVERY", "Strategy: api_json (version-first)")
        logger.verbose("DISCOVERY", f"API URL: {api_url}")
        logger.verbose("DISCOVERY", f"Version path: {version_path}")
        logger.verbose("DISCOVERY", f"Download URL path: {download_url_path}")

        # Expand environment variables in headers
        expanded_headers = {}
        for key, value in headers.items():
            if (
                isinstance(value, str)
                and value.startswith("${")
                and value.endswith("}")
            ):
                env_var = value[2:-1]
                env_value = os.environ.get(env_var)
                if not env_value:
                    logger.verbose(
                        "DISCOVERY",
                        f"Warning: Environment variable {env_var} not set",
                    )
                else:
                    expanded_headers[key] = env_value
            else:
                expanded_headers[key] = value

        # Make API request (the shared session retries transient failures)
        logger.verbose("DISCOVERY", f"Calling API: GET {api_url}")
        try:
            with make_session() as session:
                response = session.get(api_url, headers=expanded_headers, timeout=30)
        except requests.exceptions.RequestException as err:
            raise NetworkError(f"Failed to call API: {err}") from err

        if not response.ok:
            raise NetworkError(
                f"API request failed: {response.status_code} {response.reason}"
            )

        logger.verbose("DISCOVERY", f"API response: {response.status_code} OK")

        # Parse JSON response
        try:
            json_data = response.json()
        except json.JSONDecodeError as err:
            raise NetworkError(
                f"Invalid JSON response from API. Response: {response.text[:200]}"
            ) from err

        logger.debug("DISCOVERY", f"JSON response: {json.dumps(json_data, indent=2)}")

        # Extract version using JSONPath
        logger.verbose("DISCOVERY", f"Extracting version from path: {version_path}")
        try:
            version_expr = jsonpath_parse(version_path)
            version_matches = version_expr.find(json_data)

            if not version_matches:
                raise ConfigError(
                    f"Version path {version_path!r} did not match anything "
                    f"in API response"
                )

            version_str = str(version_matches[0].value)
        except Exception as err:
            if isinstance(err, ConfigError):
                raise
            raise ConfigError(
                f"Failed to extract version using path {version_path!r}: {err}"
            ) from err

        logger.verbose("DISCOVERY", f"Extracted version: {version_str}")

        version_pattern = source.get("version_pattern")
        if version_pattern:
            logger.verbose("DISCOVERY", f"Version pattern: {version_pattern}")
            version_str = _apply_version_pattern(version_pattern, version_str)
            logger.verbose("DISCOVERY", f"Version after pattern: {version_str}")

        # Extract download URL using JSONPath
        logger.verbose(
            "DISCOVERY", f"Extracting download URL from path: {download_url_path}"
        )
        try:
            url_expr = jsonpath_parse(download_url_path)
            url_matches = url_expr.find(json_data)

            if not url_matches:
                raise ConfigError(
                    f"Download URL path {download_url_path!r} did not match "
                    f"anything in API response"
                )

            download_url = str(url_matches[0].value)
        except Exception as err:
            if isinstance(err, ConfigError):
                raise
            raise ConfigError(
                f"Failed to extract download URL using path "
                f"{download_url_path!r}: {err}"
            ) from err

        logger.verbose("DISCOVERY", f"Download URL: {download_url}")

        return RemoteVersion(
            version=version_str,
            download_url=download_url,
            source="api_json",
        )

    def validate_config(self, app_config: dict[str, Any]) -> list[str]:
        """Validate api_json strategy configuration.

        Checks for required fields and correct types without making network calls.

        Args:
            app_config: The app configuration from the recipe.

        Returns:
            List of error messages (empty if valid).

        """
        errors = []
        source = app_config.get("discovery", {})

        # Check required fields
        if "api_url" not in source:
            errors.append("Missing required field: discovery.api_url")
        elif not isinstance(source["api_url"], str):
            errors.append("discovery.api_url must be a string")
        elif not source["api_url"].strip():
            errors.append("discovery.api_url cannot be empty")

        if "version_path" not in source:
            errors.append("Missing required field: discovery.version_path")
        elif not isinstance(source["version_path"], str):
            errors.append("discovery.version_path must be a string")
        elif not source["version_path"].strip():
            errors.append("discovery.version_path cannot be empty")
        else:
            # Validate JSONPath syntax
            from jsonpath_ng import parse as jsonpath_parse

            try:
                jsonpath_parse(source["version_path"])
            except Exception as err:
                errors.append(f"Invalid version_path JSONPath: {err}")

        if "download_url_path" not in source:
            errors.append("Missing required field: discovery.download_url_path")
        elif not isinstance(source["download_url_path"], str):
            errors.append("discovery.download_url_path must be a string")
        elif not source["download_url_path"].strip():
            errors.append("discovery.download_url_path cannot be empty")
        else:
            # Validate JSONPath syntax
            from jsonpath_ng import parse as jsonpath_parse

            try:
                jsonpath_parse(source["download_url_path"])
            except Exception as err:
                errors.append(f"Invalid download_url_path JSONPath: {err}")

        # Optional fields validation
        if "headers" in source and not isinstance(source["headers"], dict):
            errors.append("discovery.headers must be a dictionary")

        if "version_pattern" in source:
            if not isinstance(source["version_pattern"], str):
                errors.append("discovery.version_pattern must be a string")
            else:
                try:
                    re.compile(source["version_pattern"])
                except re.error as err:
                    errors.append(f"Invalid version_pattern regex: {err}")

        return errors

discover 

discover(app_config: dict[str, Any]) -> RemoteVersion

Discovers version and download URL from a JSON API endpoint.

Sends a GET request to the configured api_url and extracts the version and download URL using JSONPath expressions.

Parameters:

Name Type Description Default
app_config dict[str, Any]

Merged recipe configuration dict containing discovery.api_url, discovery.version_path, and discovery.download_url_path, plus optional version_pattern and headers fields.

 required

Returns:

Type Description
RemoteVersion

Discovered version, download URL, and "api_json" as
RemoteVersion

the source identifier.

Raises:

Type Description
ConfigError

On missing required configuration, when the JSONPath expressions do not match the response, or when version_pattern is invalid or does not match.
NetworkError

On API request failure.
Source code in napt/discovery/api_json.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def discover(self, app_config: dict[str, Any]) -> RemoteVersion:
    """Discovers version and download URL from a JSON API endpoint.

    Sends a GET request to the configured ``api_url`` and extracts
    the version and download URL using JSONPath expressions.

    Args:
        app_config: Merged recipe configuration dict containing
            ``discovery.api_url``, ``discovery.version_path``, and
            ``discovery.download_url_path``, plus optional
            ``version_pattern`` and ``headers`` fields.

    Returns:
        Discovered version, download URL, and ``"api_json"`` as
        the source identifier.

    Raises:
        ConfigError: On missing required configuration, when the
            JSONPath expressions do not match the response, or when
            ``version_pattern`` is invalid or does not match.
        NetworkError: On API request failure.

    """
    from napt.logging import get_global_logger

    logger = get_global_logger()
    # Validate configuration
    source = app_config.get("discovery", {})
    api_url = source.get("api_url")
    if not api_url:
        raise ConfigError(
            "api_json strategy requires 'discovery.api_url' in config"
        )

    version_path = source.get("version_path")
    if not version_path:
        raise ConfigError(
            "api_json strategy requires 'discovery.version_path' in config"
        )

    download_url_path = source.get("download_url_path")
    if not download_url_path:
        raise ConfigError(
            "api_json strategy requires 'discovery.download_url_path' in config"
        )

    # Optional configuration
    headers = source.get("headers", {})

    logger.verbose("DISCOVERY", "Strategy: api_json (version-first)")
    logger.verbose("DISCOVERY", f"API URL: {api_url}")
    logger.verbose("DISCOVERY", f"Version path: {version_path}")
    logger.verbose("DISCOVERY", f"Download URL path: {download_url_path}")

    # Expand environment variables in headers
    expanded_headers = {}
    for key, value in headers.items():
        if (
            isinstance(value, str)
            and value.startswith("${")
            and value.endswith("}")
        ):
            env_var = value[2:-1]
            env_value = os.environ.get(env_var)
            if not env_value:
                logger.verbose(
                    "DISCOVERY",
                    f"Warning: Environment variable {env_var} not set",
                )
            else:
                expanded_headers[key] = env_value
        else:
            expanded_headers[key] = value

    # Make API request (the shared session retries transient failures)
    logger.verbose("DISCOVERY", f"Calling API: GET {api_url}")
    try:
        with make_session() as session:
            response = session.get(api_url, headers=expanded_headers, timeout=30)
    except requests.exceptions.RequestException as err:
        raise NetworkError(f"Failed to call API: {err}") from err

    if not response.ok:
        raise NetworkError(
            f"API request failed: {response.status_code} {response.reason}"
        )

    logger.verbose("DISCOVERY", f"API response: {response.status_code} OK")

    # Parse JSON response
    try:
        json_data = response.json()
    except json.JSONDecodeError as err:
        raise NetworkError(
            f"Invalid JSON response from API. Response: {response.text[:200]}"
        ) from err

    logger.debug("DISCOVERY", f"JSON response: {json.dumps(json_data, indent=2)}")

    # Extract version using JSONPath
    logger.verbose("DISCOVERY", f"Extracting version from path: {version_path}")
    try:
        version_expr = jsonpath_parse(version_path)
        version_matches = version_expr.find(json_data)

        if not version_matches:
            raise ConfigError(
                f"Version path {version_path!r} did not match anything "
                f"in API response"
            )

        version_str = str(version_matches[0].value)
    except Exception as err:
        if isinstance(err, ConfigError):
            raise
        raise ConfigError(
            f"Failed to extract version using path {version_path!r}: {err}"
        ) from err

    logger.verbose("DISCOVERY", f"Extracted version: {version_str}")

    version_pattern = source.get("version_pattern")
    if version_pattern:
        logger.verbose("DISCOVERY", f"Version pattern: {version_pattern}")
        version_str = _apply_version_pattern(version_pattern, version_str)
        logger.verbose("DISCOVERY", f"Version after pattern: {version_str}")

    # Extract download URL using JSONPath
    logger.verbose(
        "DISCOVERY", f"Extracting download URL from path: {download_url_path}"
    )
    try:
        url_expr = jsonpath_parse(download_url_path)
        url_matches = url_expr.find(json_data)

        if not url_matches:
            raise ConfigError(
                f"Download URL path {download_url_path!r} did not match "
                f"anything in API response"
            )

        download_url = str(url_matches[0].value)
    except Exception as err:
        if isinstance(err, ConfigError):
            raise
        raise ConfigError(
            f"Failed to extract download URL using path "
            f"{download_url_path!r}: {err}"
        ) from err

    logger.verbose("DISCOVERY", f"Download URL: {download_url}")

    return RemoteVersion(
        version=version_str,
        download_url=download_url,
        source="api_json",
    )

validate_config 

validate_config(app_config: dict[str, Any]) -> list[str]

Validate api_json strategy configuration.

Checks for required fields and correct types without making network calls.

Parameters:

Name Type Description Default
app_config dict[str, Any]

The app configuration from the recipe.

 required

Returns:

Type Description
list[str]

List of error messages (empty if valid).
Source code in napt/discovery/api_json.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def validate_config(self, app_config: dict[str, Any]) -> list[str]:
    """Validate api_json strategy configuration.

    Checks for required fields and correct types without making network calls.

    Args:
        app_config: The app configuration from the recipe.

    Returns:
        List of error messages (empty if valid).

    """
    errors = []
    source = app_config.get("discovery", {})

    # Check required fields
    if "api_url" not in source:
        errors.append("Missing required field: discovery.api_url")
    elif not isinstance(source["api_url"], str):
        errors.append("discovery.api_url must be a string")
    elif not source["api_url"].strip():
        errors.append("discovery.api_url cannot be empty")

    if "version_path" not in source:
        errors.append("Missing required field: discovery.version_path")
    elif not isinstance(source["version_path"], str):
        errors.append("discovery.version_path must be a string")
    elif not source["version_path"].strip():
        errors.append("discovery.version_path cannot be empty")
    else:
        # Validate JSONPath syntax
        from jsonpath_ng import parse as jsonpath_parse

        try:
            jsonpath_parse(source["version_path"])
        except Exception as err:
            errors.append(f"Invalid version_path JSONPath: {err}")

    if "download_url_path" not in source:
        errors.append("Missing required field: discovery.download_url_path")
    elif not isinstance(source["download_url_path"], str):
        errors.append("discovery.download_url_path must be a string")
    elif not source["download_url_path"].strip():
        errors.append("discovery.download_url_path cannot be empty")
    else:
        # Validate JSONPath syntax
        from jsonpath_ng import parse as jsonpath_parse

        try:
            jsonpath_parse(source["download_url_path"])
        except Exception as err:
            errors.append(f"Invalid download_url_path JSONPath: {err}")

    # Optional fields validation
    if "headers" in source and not isinstance(source["headers"], dict):
        errors.append("discovery.headers must be a dictionary")

    if "version_pattern" in source:
        if not isinstance(source["version_pattern"], str):
            errors.append("discovery.version_pattern must be a string")
        else:
            try:
                re.compile(source["version_pattern"])
            except re.error as err:
                errors.append(f"Invalid version_pattern regex: {err}")

    return errors