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

96.97

/src/python/pants/bsp/util_rules/compile.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
import time
import uuid
from collections import defaultdict
from collections.abc import Sequence
from dataclasses import dataclass
from typing import TypeVar

from pants.bsp.context import BSPContext
from pants.bsp.protocol import BSPHandlerMapping
from pants.bsp.spec.base import StatusCode, TaskId
from pants.bsp.spec.compile import CompileParams, CompileReport, CompileResult, CompileTask
from pants.bsp.spec.task import TaskFinishParams, TaskStartParams
from pants.bsp.util_rules.targets import (
    BSPBuildTargetInternal,
    BSPCompileRequest,
    BSPCompileResult,
    bsp_compile,
    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 CompileRequestHandlerMapping(BSPHandlerMapping):
    method_name = "buildTarget/compile"
    request_type = CompileParams
    response_type = CompileResult


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

    # A unique identifier generated by the client to identify this request.
    # The server may include this id in triggered notifications or responses.
    origin_id: str | None = None

    # Optional arguments to the compilation process.
    arguments: tuple[str, ...] | None = ()


@rule
async def compile_bsp_target(
    request: CompileOneBSPTargetRequest,
    bsp_context: BSPContext,
    union_membership: UnionMembership,
) -> BSPCompileResult:
    targets = await resolve_bsp_build_target_addresses(request.bsp_target, **implicitly())
    compile_request_types: Sequence[type[BSPCompileRequest]] = union_membership.get(
        BSPCompileRequest
    )
    field_sets_by_request_type: dict[type[BSPCompileRequest], set[FieldSet]] = defaultdict(set)
    for target in targets:
        for compile_request_type in compile_request_types:
            field_set_type = compile_request_type.field_set_type
            if field_set_type.is_applicable(target):
                field_set = field_set_type.create(target)
                field_sets_by_request_type[compile_request_type].add(field_set)

    task_id = TaskId(
        id=uuid.uuid4().hex, parents=((request.origin_id,) if request.origin_id else None)
    )
    message = f"Compilation of {request.bsp_target.bsp_target_id.uri}"

    bsp_context.notify_client(
        TaskStartParams(
            task_id=task_id,
            event_time=int(time.time() * 1000),
            message=message,
            data=CompileTask(target=request.bsp_target.bsp_target_id),
        )
    )

    compile_results = await concurrently(
        bsp_compile(
            **implicitly(
                {
                    compile_request_type(
                        bsp_target=request.bsp_target, field_sets=tuple(field_sets), task_id=task_id
                    ): BSPCompileRequest
                }
            )
        )
        for compile_request_type, field_sets in field_sets_by_request_type.items()
    )

    status = StatusCode.OK
    if any(r.status != StatusCode.OK for r in compile_results):
        status = StatusCode.ERROR

    bsp_context.notify_client(
        TaskFinishParams(
            task_id=task_id,
            event_time=int(time.time() * 1000),
            message=message,
            status=status,
            data=CompileReport(
                target=request.bsp_target.bsp_target_id,
                origin_id=request.origin_id,
                errors=0,
                warnings=0,
            ),
        )
    )

    output_digest = await merge_digests(MergeDigests([r.output_digest for r in compile_results]))

    return BSPCompileResult(
        status=status,
        output_digest=output_digest,
    )


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

    compile_results = await concurrently(
        compile_bsp_target(
            CompileOneBSPTargetRequest(
                bsp_target=bsp_target,
                origin_id=request.origin_id,
                arguments=request.arguments,
            ),
            **implicitly(),
        )
        for bsp_target in bsp_targets
    )

    output_digest = await merge_digests(MergeDigests([r.output_digest for r in compile_results]))
    if output_digest != EMPTY_DIGEST:
        workspace.write_digest(output_digest, path_prefix=".pants.d/bsp")

    status_code = StatusCode.OK
    if any(r.status != StatusCode.OK for r in compile_results):
        status_code = StatusCode.ERROR

    return CompileResult(
        origin_id=request.origin_id,
        status_code=status_code.value,
    )


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

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	import time	2✔
7	import uuid	2✔
8	from collections import defaultdict	2✔
9	from collections.abc import Sequence	2✔
10	from dataclasses import dataclass	2✔
11	from typing import TypeVar	2✔
12
13	from pants.bsp.context import BSPContext	2✔
14	from pants.bsp.protocol import BSPHandlerMapping	2✔
15	from pants.bsp.spec.base import StatusCode, TaskId	2✔
16	from pants.bsp.spec.compile import CompileParams, CompileReport, CompileResult, CompileTask	2✔
17	from pants.bsp.spec.task import TaskFinishParams, TaskStartParams	2✔
18	from pants.bsp.util_rules.targets import (	2✔
19	BSPBuildTargetInternal,
20	BSPCompileRequest,
21	BSPCompileResult,
22	bsp_compile,
23	resolve_bsp_build_target_addresses,
24	resolve_bsp_build_target_identifier,
25	)
26	from pants.engine.fs import Workspace	2✔
27	from pants.engine.internals.native_engine import EMPTY_DIGEST, MergeDigests	2✔
28	from pants.engine.internals.selectors import concurrently	2✔
29	from pants.engine.intrinsics import merge_digests	2✔
30	from pants.engine.rules import _uncacheable_rule, collect_rules, implicitly, rule	2✔
31	from pants.engine.target import FieldSet	2✔
32	from pants.engine.unions import UnionMembership, UnionRule	2✔
33
34	_logger = logging.getLogger(__name__)	2✔
35
36	_FS = TypeVar("_FS", bound=FieldSet)	2✔
37
38
39	class CompileRequestHandlerMapping(BSPHandlerMapping):	2✔
40	method_name = "buildTarget/compile"	2✔
41	request_type = CompileParams	2✔
42	response_type = CompileResult	2✔
43
44
45	@dataclass(frozen=True)	2✔
46	class CompileOneBSPTargetRequest:	2✔
47	bsp_target: BSPBuildTargetInternal	2✔
48
49	# A unique identifier generated by the client to identify this request.
50	# The server may include this id in triggered notifications or responses.
51	origin_id: str \| None = None	2✔
52
53	# Optional arguments to the compilation process.
54	arguments: tuple[str, ...] \| None = ()	2✔
55
56
57	@rule	2✔
58	async def compile_bsp_target(	2✔
59	request: CompileOneBSPTargetRequest,
60	bsp_context: BSPContext,
61	union_membership: UnionMembership,
62	) -> BSPCompileResult:
63	targets = await resolve_bsp_build_target_addresses(request.bsp_target, **implicitly())	1✔
64	compile_request_types: Sequence[type[BSPCompileRequest]] = union_membership.get(	1✔
65	BSPCompileRequest
66	)
67	field_sets_by_request_type: dict[type[BSPCompileRequest], set[FieldSet]] = defaultdict(set)	1✔
68	for target in targets:	1✔
69	for compile_request_type in compile_request_types:	1✔
70	field_set_type = compile_request_type.field_set_type	1✔
71	if field_set_type.is_applicable(target):	1✔
72	field_set = field_set_type.create(target)	1✔
73	field_sets_by_request_type[compile_request_type].add(field_set)	1✔
74
75	task_id = TaskId(	1✔
76	id=uuid.uuid4().hex, parents=((request.origin_id,) if request.origin_id else None)
77	)
78	message = f"Compilation of {request.bsp_target.bsp_target_id.uri}"	1✔
79
80	bsp_context.notify_client(	1✔
81	TaskStartParams(
82	task_id=task_id,
83	event_time=int(time.time() * 1000),
84	message=message,
85	data=CompileTask(target=request.bsp_target.bsp_target_id),
86	)
87	)
88
89	compile_results = await concurrently(	1✔
90	bsp_compile(
91	**implicitly(
92	{
93	compile_request_type(
94	bsp_target=request.bsp_target, field_sets=tuple(field_sets), task_id=task_id
95	): BSPCompileRequest
96	}
97	)
98	)
99	for compile_request_type, field_sets in field_sets_by_request_type.items()
100	)
101
102	status = StatusCode.OK	1✔
103	if any(r.status != StatusCode.OK for r in compile_results):	1✔
104	status = StatusCode.ERROR	×
105
106	bsp_context.notify_client(	1✔
107	TaskFinishParams(
108	task_id=task_id,
109	event_time=int(time.time() * 1000),
110	message=message,
111	status=status,
112	data=CompileReport(
113	target=request.bsp_target.bsp_target_id,
114	origin_id=request.origin_id,
115	errors=0,
116	warnings=0,
117	),
118	)
119	)
120
121	output_digest = await merge_digests(MergeDigests([r.output_digest for r in compile_results]))	1✔
122
123	return BSPCompileResult(	1✔
124	status=status,
125	output_digest=output_digest,
126	)
127
128
129	@_uncacheable_rule	2✔
130	async def bsp_compile_request(	2✔
131	request: CompileParams,
132	workspace: Workspace,
133	) -> CompileResult:
134	bsp_targets = await concurrently(	1✔
135	resolve_bsp_build_target_identifier(bsp_target_id, **implicitly())
136	for bsp_target_id in request.targets
137	)
138
139	compile_results = await concurrently(	1✔
140	compile_bsp_target(
141	CompileOneBSPTargetRequest(
142	bsp_target=bsp_target,
143	origin_id=request.origin_id,
144	arguments=request.arguments,
145	),
146	**implicitly(),
147	)
148	for bsp_target in bsp_targets
149	)
150
151	output_digest = await merge_digests(MergeDigests([r.output_digest for r in compile_results]))	1✔
152	if output_digest != EMPTY_DIGEST:	1✔
153	workspace.write_digest(output_digest, path_prefix=".pants.d/bsp")	1✔
154
155	status_code = StatusCode.OK	1✔
156	if any(r.status != StatusCode.OK for r in compile_results):	1✔
157	status_code = StatusCode.ERROR	×
158
159	return CompileResult(	1✔
160	origin_id=request.origin_id,
161	status_code=status_code.value,
162	)
163
164
165	def rules():	2✔
166	return (	2✔
167	*collect_rules(),
168	UnionRule(BSPHandlerMapping, CompileRequestHandlerMapping),
169	)

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