25441711719

Committed 06 May 2026 02:31PM UTC coverage: 92.915%. Remained the same

Build # 25441711719

Build Type

push

github

Committed by

web-flow

Commit Message

use sha pin (with comment) format for generated actions (#23312)

Per the GitHub Action best practices we recently enabled at #23249, we
should pin each action to a SHA so that the reference is actually
immutable.

This will -- I hope -- knock out a large chunk of the 421 alerts we
currently get from zizmor. The next followup would then be upgrades and
harmonizing the generated and none-generated pins.

Notice: This idea was suggested by Claude while going over pinact output
and I was surprised to see that post processing the yaml wasn't too
gross.

Coverage Stats

92206 of 99237 relevant lines covered (92.91%)

4.04 hits per line

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

100.0

/src/python/pants/bsp/util_rules/resources.py

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

import logging
from collections import defaultdict
from collections.abc import Sequence
from dataclasses import dataclass
from typing import TypeVar

from pants.bsp.protocol import BSPHandlerMapping
from pants.bsp.spec.resources import ResourcesItem, ResourcesParams, ResourcesResult
from pants.bsp.util_rules.targets import (
    BSPBuildTargetInternal,
    BSPResourcesRequest,
    BSPResourcesResult,
    get_bsp_resources,
    resolve_bsp_build_target_addresses,
    resolve_bsp_build_target_identifier,
)
from pants.engine.fs import Workspace
from pants.engine.internals.native_engine import EMPTY_DIGEST, MergeDigests
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import merge_digests
from pants.engine.rules import _uncacheable_rule, collect_rules, implicitly, rule
from pants.engine.target import FieldSet
from pants.engine.unions import UnionMembership, UnionRule

_logger = logging.getLogger(__name__)

_FS = TypeVar("_FS", bound=FieldSet)


class ResourcesRequestHandlerMapping(BSPHandlerMapping):
    method_name = "buildTarget/resources"
    request_type = ResourcesParams
    response_type = ResourcesResult


@dataclass(frozen=True)
class ResourcesForOneBSPTargetRequest:
    bsp_target: BSPBuildTargetInternal


@rule
async def resources_bsp_target(
    request: ResourcesForOneBSPTargetRequest,
    union_membership: UnionMembership,
) -> BSPResourcesResult:
    targets = await resolve_bsp_build_target_addresses(request.bsp_target, **implicitly())
    resources_request_types: Sequence[type[BSPResourcesRequest]] = union_membership.get(
        BSPResourcesRequest
    )
    field_sets_by_request_type: dict[type[BSPResourcesRequest], set[FieldSet]] = defaultdict(set)
    for target in targets:
        for resources_request_type in resources_request_types:
            field_set_type = resources_request_type.field_set_type
            if field_set_type.is_applicable(target):
                field_set = field_set_type.create(target)
                field_sets_by_request_type[resources_request_type].add(field_set)

    resources_results = await concurrently(
        get_bsp_resources(
            **implicitly(
                {
                    resources_request_type(
                        bsp_target=request.bsp_target, field_sets=tuple(field_sets)
                    ): BSPResourcesRequest
                }
            )
        )
        for resources_request_type, field_sets in field_sets_by_request_type.items()
    )

    resources = tuple(sorted({resource for rr in resources_results for resource in rr.resources}))

    output_digest = await merge_digests(
        MergeDigests([rr.output_digest for rr in resources_results])
    )

    return BSPResourcesResult(
        resources=resources,
        output_digest=output_digest,
    )


@_uncacheable_rule
async def bsp_resources_request(
    request: ResourcesParams,
    workspace: Workspace,
) -> ResourcesResult:
    bsp_targets = await concurrently(
        resolve_bsp_build_target_identifier(bsp_target_id, **implicitly())
        for bsp_target_id in request.targets
    )

    resources_results = await concurrently(
        resources_bsp_target(
            ResourcesForOneBSPTargetRequest(
                bsp_target=bsp_target,
            ),
            **implicitly(),
        )
        for bsp_target in bsp_targets
    )

    # TODO: Need to determine how resources are expected to be exposed. Directories? Individual files?
    # Initially, it looks like loose directories.
    output_digest = await merge_digests(MergeDigests([r.output_digest for r in resources_results]))
    if output_digest != EMPTY_DIGEST:
        workspace.write_digest(output_digest, path_prefix=".pants.d/bsp")

    return ResourcesResult(
        tuple(
            ResourcesItem(
                target,
                rr.resources,
            )
            for target, rr in zip(request.targets, resources_results)
        )
    )


def rules():
    return (
        *collect_rules(),
        UnionRule(BSPHandlerMapping, ResourcesRequestHandlerMapping),
    )

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	2✔
4
5	import logging	2✔
6	from collections import defaultdict	2✔
7	from collections.abc import Sequence	2✔
8	from dataclasses import dataclass	2✔
9	from typing import TypeVar	2✔
10
11	from pants.bsp.protocol import BSPHandlerMapping	2✔
12	from pants.bsp.spec.resources import ResourcesItem, ResourcesParams, ResourcesResult	2✔
13	from pants.bsp.util_rules.targets import (	2✔
14	BSPBuildTargetInternal,
15	BSPResourcesRequest,
16	BSPResourcesResult,
17	get_bsp_resources,
18	resolve_bsp_build_target_addresses,
19	resolve_bsp_build_target_identifier,
20	)
21	from pants.engine.fs import Workspace	2✔
22	from pants.engine.internals.native_engine import EMPTY_DIGEST, MergeDigests	2✔
23	from pants.engine.internals.selectors import concurrently	2✔
24	from pants.engine.intrinsics import merge_digests	2✔
25	from pants.engine.rules import _uncacheable_rule, collect_rules, implicitly, rule	2✔
26	from pants.engine.target import FieldSet	2✔
27	from pants.engine.unions import UnionMembership, UnionRule	2✔
28
29	_logger = logging.getLogger(__name__)	2✔
30
31	_FS = TypeVar("_FS", bound=FieldSet)	2✔
32
33
34	class ResourcesRequestHandlerMapping(BSPHandlerMapping):	2✔
35	method_name = "buildTarget/resources"	2✔
36	request_type = ResourcesParams	2✔
37	response_type = ResourcesResult	2✔
38
39
40	@dataclass(frozen=True)	2✔
41	class ResourcesForOneBSPTargetRequest:	2✔
42	bsp_target: BSPBuildTargetInternal	2✔
43
44
45	@rule	2✔
46	async def resources_bsp_target(	2✔
47	request: ResourcesForOneBSPTargetRequest,
48	union_membership: UnionMembership,
49	) -> BSPResourcesResult:
50	targets = await resolve_bsp_build_target_addresses(request.bsp_target, **implicitly())	1✔
51	resources_request_types: Sequence[type[BSPResourcesRequest]] = union_membership.get(	1✔
52	BSPResourcesRequest
53	)
54	field_sets_by_request_type: dict[type[BSPResourcesRequest], set[FieldSet]] = defaultdict(set)	1✔
55	for target in targets:	1✔
56	for resources_request_type in resources_request_types:	1✔
57	field_set_type = resources_request_type.field_set_type	1✔
58	if field_set_type.is_applicable(target):	1✔
59	field_set = field_set_type.create(target)	1✔
60	field_sets_by_request_type[resources_request_type].add(field_set)	1✔
61
62	resources_results = await concurrently(	1✔
63	get_bsp_resources(
64	**implicitly(
65	{
66	resources_request_type(
67	bsp_target=request.bsp_target, field_sets=tuple(field_sets)
68	): BSPResourcesRequest
69	}
70	)
71	)
72	for resources_request_type, field_sets in field_sets_by_request_type.items()
73	)
74
75	resources = tuple(sorted({resource for rr in resources_results for resource in rr.resources}))	1✔
76
77	output_digest = await merge_digests(	1✔
78	MergeDigests([rr.output_digest for rr in resources_results])
79	)
80
81	return BSPResourcesResult(	1✔
82	resources=resources,
83	output_digest=output_digest,
84	)
85
86
87	@_uncacheable_rule	2✔
88	async def bsp_resources_request(	2✔
89	request: ResourcesParams,
90	workspace: Workspace,
91	) -> ResourcesResult:
92	bsp_targets = await concurrently(	1✔
93	resolve_bsp_build_target_identifier(bsp_target_id, **implicitly())
94	for bsp_target_id in request.targets
95	)
96
97	resources_results = await concurrently(	1✔
98	resources_bsp_target(
99	ResourcesForOneBSPTargetRequest(
100	bsp_target=bsp_target,
101	),
102	**implicitly(),
103	)
104	for bsp_target in bsp_targets
105	)
106
107	# TODO: Need to determine how resources are expected to be exposed. Directories? Individual files?
108	# Initially, it looks like loose directories.
109	output_digest = await merge_digests(MergeDigests([r.output_digest for r in resources_results]))	1✔
110	if output_digest != EMPTY_DIGEST:	1✔
111	workspace.write_digest(output_digest, path_prefix=".pants.d/bsp")	1✔
112
113	return ResourcesResult(	1✔
114	tuple(
115	ResourcesItem(
116	target,
117	rr.resources,
118	)
119	for target, rr in zip(request.targets, resources_results)
120	)
121	)
122
123
124	def rules():	2✔
125	return (	2✔
126	*collect_rules(),
127	UnionRule(BSPHandlerMapping, ResourcesRequestHandlerMapping),
128	)

pantsbuild / pants / 25441711719

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