19250292619

Committed 11 Nov 2025 12:09AM UTC coverage: 77.865% (-2.4%) from 80.298%

Build # 19250292619

Build Type

push

github

Committed by

web-flow

Commit Message

flag non-runnable targets used with `code_quality_tool` (#22875)

Run Details

2 of 5 new or added lines in 2 files covered. (40.0%)

1487 existing lines in 72 files now uncovered.

71448 of 91759 relevant lines covered (77.86%)

3.22 hits per line

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

63.41

/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	11✔
5	from fnmatch import fnmatch	11✔
6	from typing import ClassVar	11✔
7	from urllib.parse import urlparse	11✔
8
9	from pants.engine.fs import Digest, DownloadFile, NativeDownloadFile	11✔
10	from pants.engine.internals.native_engine import FileDigest	11✔
11	from pants.engine.intrinsics import download_file as native_download_file	11✔
12	from pants.engine.rules import collect_rules, implicitly, rule	11✔
13	from pants.engine.unions import UnionMembership, union	11✔
14	from pants.option.global_options import GlobalOptions	11✔
15	from pants.util.strutil import bullet_list, softwrap	11✔
16
17
18	@union	11✔
19	@dataclass(frozen=True)	11✔
20	class URLDownloadHandler:	11✔
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	11✔
47	"""The scheme to match (e.g. 'ftp' or 's3') or `None` to match all schemes.	11✔
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	11✔
54	"""The authority to match (e.g. 'pantsbuild.org' or 's3.amazonaws.com') or `None` to match all authorities.	11✔
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	11✔
64	expected_digest: FileDigest	11✔
65
66
67	@rule(polymorphic=True)	11✔
68	async def download_file_using_handler(handler: URLDownloadHandler) -> Digest:	11✔
69	raise NotImplementedError()	×
70
71
72	@rule	11✔
73	async def download_file(	11✔
74	request: DownloadFile,
75	union_membership: UnionMembership,
76	global_options: GlobalOptions,
77	) -> Digest:
UNCOV 78	parsed_url = urlparse(request.url)	×
79
UNCOV 80	handlers = union_membership.get(URLDownloadHandler)	×
UNCOV 81	matched_handlers = []	×
UNCOV 82	for handler in handlers:	×
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
UNCOV 92	if len(matched_handlers) > 1:	×
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	)
UNCOV 103	if len(matched_handlers) == 1:	×
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
UNCOV 109	return await native_download_file(	×
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():	11✔
120	return collect_rules()	11✔

pantsbuild / pants / 19250292619

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