22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.49

/src/python/pants/engine/download_file.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from dataclasses import dataclass
from fnmatch import fnmatch
from typing import ClassVar
from urllib.parse import urlparse

from pants.engine.fs import Digest, DownloadFile, NativeDownloadFile
from pants.engine.internals.native_engine import FileDigest
from pants.engine.intrinsics import download_file as native_download_file
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.unions import UnionMembership, union
from pants.option.global_options import GlobalOptions
from pants.util.strutil import bullet_list, softwrap


@union
@dataclass(frozen=True)
class URLDownloadHandler:
    """Union base for custom URL handler.

    To register a custom URL handler:
    - Subclass this class and declare one or both of the ClassVars.
    - Declare a rule that takes in your class type and returns a `Digest`.
    - Register your union member in your `rules()`: `UnionRule(URLDownloadHandler, YourClass)`.

    Example:

        class S3DownloadHandler(URLDownloadHandler):
            match_scheme = "s3"

        @rule
        async def download_s3_file(request: S3DownloadHandler) -> Digest:
            # Lookup auth tokens, etc...
            # Ideally, download the file using `NativeDownloadFile()`
            return digest

        def rules():
            return [
                *collect_rules(),
                UnionRule(URLDownloadHandler, S3DownloadHandler),
            ]
    """

    match_scheme: ClassVar[str | None] = None
    """The scheme to match (e.g. 'ftp' or 's3') or `None` to match all schemes.

    The scheme is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
    information.
    """

    match_authority: ClassVar[str | None] = None
    """The authority to match (e.g. 'pantsbuild.org' or 's3.amazonaws.com') or `None` to match all authorities.

    The authority is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
    information.

    Note that the authority matches userinfo (e.g. 'me@pantsbuild.org' or 'me:password@pantsbuild.org')
    as well as port (e.g. 'pantsbuild.org:80').
    """

    url: str
    expected_digest: FileDigest


@rule(polymorphic=True)
async def download_file_using_handler(handler: URLDownloadHandler) -> Digest:
    raise NotImplementedError()


@rule
async def download_file(
    request: DownloadFile,
    union_membership: UnionMembership,
    global_options: GlobalOptions,
) -> Digest:
    parsed_url = urlparse(request.url)

    handlers = union_membership.get(URLDownloadHandler)
    matched_handlers = []
    for handler in handlers:
        matches_scheme = handler.match_scheme is None or fnmatch(
            parsed_url.scheme, handler.match_scheme
        )
        matches_authority = handler.match_authority is None or fnmatch(
            parsed_url.netloc, handler.match_authority
        )
        if matches_scheme and matches_authority:
            matched_handlers.append(handler)

    if len(matched_handlers) > 1:
        raise Exception(
            softwrap(
                f"""
                Too many registered URL handlers match the URL '{request.url}'.

                Matched handlers:
                {bullet_list(map(str, handlers))}
                """
            )
        )
    if len(matched_handlers) == 1:
        handler = matched_handlers[0]
        return await download_file_using_handler(
            **implicitly({handler(request.url, request.expected_digest): URLDownloadHandler})
        )

    return await native_download_file(
        NativeDownloadFile(
            request.url,
            request.expected_digest,
            retry_delay_duration=global_options.file_downloads_retry_delay,
            max_attempts=global_options.file_downloads_max_attempts,
        ),
    )


def rules():
    return collect_rules()

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from dataclasses import dataclass	4✔
5	from fnmatch import fnmatch	4✔
6	from typing import ClassVar	4✔
7	from urllib.parse import urlparse	4✔
8
9	from pants.engine.fs import Digest, DownloadFile, NativeDownloadFile	4✔
10	from pants.engine.internals.native_engine import FileDigest	4✔
11	from pants.engine.intrinsics import download_file as native_download_file	4✔
12	from pants.engine.rules import collect_rules, implicitly, rule	4✔
13	from pants.engine.unions import UnionMembership, union	4✔
14	from pants.option.global_options import GlobalOptions	4✔
15	from pants.util.strutil import bullet_list, softwrap	4✔
16
17
18	@union	4✔
19	@dataclass(frozen=True)	4✔
20	class URLDownloadHandler:	4✔
21	"""Union base for custom URL handler.
22
23	To register a custom URL handler:
24	- Subclass this class and declare one or both of the ClassVars.
25	- Declare a rule that takes in your class type and returns a `Digest`.
26	- Register your union member in your `rules()`: `UnionRule(URLDownloadHandler, YourClass)`.
27
28	Example:
29
30	class S3DownloadHandler(URLDownloadHandler):
31	match_scheme = "s3"
32
33	@rule
34	async def download_s3_file(request: S3DownloadHandler) -> Digest:
35	# Lookup auth tokens, etc...
36	# Ideally, download the file using `NativeDownloadFile()`
37	return digest
38
39	def rules():
40	return [
41	*collect_rules(),
42	UnionRule(URLDownloadHandler, S3DownloadHandler),
43	]
44	"""
45
46	match_scheme: ClassVar[str \| None] = None	4✔
47	"""The scheme to match (e.g. 'ftp' or 's3') or `None` to match all schemes.	4✔
48
49	The scheme is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
50	information.
51	"""
52
53	match_authority: ClassVar[str \| None] = None	4✔
54	"""The authority to match (e.g. 'pantsbuild.org' or 's3.amazonaws.com') or `None` to match all authorities.	4✔
55
56	The authority is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
57	information.
58
59	Note that the authority matches userinfo (e.g. 'me@pantsbuild.org' or 'me:password@pantsbuild.org')
60	as well as port (e.g. 'pantsbuild.org:80').
61	"""
62
63	url: str
64	expected_digest: FileDigest
65
66
67	@rule(polymorphic=True)	4✔
68	async def download_file_using_handler(handler: URLDownloadHandler) -> Digest:	4✔
69	raise NotImplementedError()	×
70
71
72	@rule	4✔
73	async def download_file(	4✔
74	request: DownloadFile,
75	union_membership: UnionMembership,
76	global_options: GlobalOptions,
77	) -> Digest:
78	parsed_url = urlparse(request.url)	4✔
79
80	handlers = union_membership.get(URLDownloadHandler)	4✔
81	matched_handlers = []	4✔
82	for handler in handlers:	4✔
UNCOV 83	matches_scheme = handler.match_scheme is None or fnmatch(	×
84	parsed_url.scheme, handler.match_scheme
85	)
UNCOV 86	matches_authority = handler.match_authority is None or fnmatch(	×
87	parsed_url.netloc, handler.match_authority
88	)
UNCOV 89	if matches_scheme and matches_authority:	×
UNCOV 90	matched_handlers.append(handler)	×
91
92	if len(matched_handlers) > 1:	4✔
UNCOV 93	raise Exception(	×
94	softwrap(
95	f"""
96	Too many registered URL handlers match the URL '{request.url}'.
97
98	Matched handlers:
99	{bullet_list(map(str, handlers))}
100	"""
101	)
102	)
103	if len(matched_handlers) == 1:	4✔
UNCOV 104	handler = matched_handlers[0]	×
UNCOV 105	return await download_file_using_handler(	×
106	**implicitly({handler(request.url, request.expected_digest): URLDownloadHandler})
107	)
108
109	return await native_download_file(	4✔
110	NativeDownloadFile(
111	request.url,
112	request.expected_digest,
113	retry_delay_duration=global_options.file_downloads_retry_delay,
114	max_attempts=global_options.file_downloads_max_attempts,
115	),
116	)
117
118
119	def rules():	4✔
120	return collect_rules()	4✔

pantsbuild / pants / 22285099215

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous