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/backend/openapi/dependency_inference.py

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

import contextlib
import json
import os.path
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

import yaml

from pants.backend.openapi.target_types import (
    OPENAPI_FILE_EXTENSIONS,
    OpenApiDocumentDependenciesField,
    OpenApiDocumentField,
    OpenApiSourceDependenciesField,
    OpenApiSourceField,
)
from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.base.specs import FileLiteralSpec, RawSpecs
from pants.engine.fs import Digest
from pants.engine.internals.graph import (
    determine_explicitly_provided_dependencies,
    hydrate_sources,
    resolve_targets,
)
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import get_digest_contents
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import (
    DependenciesRequest,
    FieldSet,
    HydrateSourcesRequest,
    InferDependenciesRequest,
    InferredDependencies,
)
from pants.engine.unions import UnionRule
from pants.util.frozendict import FrozenDict


@dataclass(frozen=True)
class ParseOpenApiSources:
    sources_digest: Digest
    paths: tuple[str, ...]


@dataclass(frozen=True)
class OpenApiDependencies:
    dependencies: FrozenDict[str, frozenset[str]]


@rule
async def parse_openapi_sources(request: ParseOpenApiSources) -> OpenApiDependencies:
    digest_contents = await get_digest_contents(request.sources_digest)
    dependencies: dict[str, frozenset[str]] = {}

    for digest_content in digest_contents:
        spec = None

        if digest_content.path.endswith(".json"):
            with contextlib.suppress(json.JSONDecodeError):
                spec = json.loads(digest_content.content)
        elif digest_content.path.endswith(".yaml") or digest_content.path.endswith(".yml"):
            with contextlib.suppress(yaml.YAMLError):
                spec = yaml.safe_load(digest_content.content)

        if not spec or not isinstance(spec, dict):
            dependencies[digest_content.path] = frozenset()
            continue

        dependencies[digest_content.path] = _find_local_refs(digest_content.path, spec)

    return OpenApiDependencies(dependencies=FrozenDict(dependencies))


def _find_local_refs(path: str, d: Mapping[str, Any]) -> frozenset[str]:
    local_refs: set[str] = set()

    for k, v in d.items():
        if isinstance(v, dict):
            local_refs.update(_find_local_refs(path, v))
        elif k == "$ref" and isinstance(v, str):
            # https://swagger.io/specification/#reference-object
            # https://datatracker.ietf.org/doc/html/draft-pbryan-zyp-json-ref-03
            v = v.split("#", 1)[0]

            if any(v.endswith(ext) for ext in OPENAPI_FILE_EXTENSIONS) and "://" not in v:
                # Resolution is performed relative to the referring document.
                normalized = os.path.normpath(os.path.join(os.path.dirname(path), v))

                if not normalized.startswith("../"):
                    local_refs.add(normalized)

    return frozenset(local_refs)


# -----------------------------------------------------------------------------------------------
# `openapi_document` dependency inference
# -----------------------------------------------------------------------------------------------


@dataclass(frozen=True)
class OpenApiDocumentDependenciesInferenceFieldSet(FieldSet):
    required_fields = (OpenApiDocumentField, OpenApiDocumentDependenciesField)

    sources: OpenApiDocumentField
    dependencies: OpenApiDocumentDependenciesField


class InferOpenApiDocumentDependenciesRequest(InferDependenciesRequest):
    infer_from = OpenApiDocumentDependenciesInferenceFieldSet


@rule
async def infer_openapi_document_dependencies(
    request: InferOpenApiDocumentDependenciesRequest,
) -> InferredDependencies:
    explicitly_provided_deps, hydrated_sources = await concurrently(
        determine_explicitly_provided_dependencies(
            **implicitly(DependenciesRequest(request.field_set.dependencies))
        ),
        hydrate_sources(HydrateSourcesRequest(request.field_set.sources), **implicitly()),
    )
    candidate_targets = await resolve_targets(
        **implicitly(
            RawSpecs(
                file_literals=(FileLiteralSpec(*hydrated_sources.snapshot.files),),
                description_of_origin="the `openapi_document` dependency inference",
            )
        )
    )

    addresses = frozenset(
        [target.address for target in candidate_targets if target.has_field(OpenApiSourceField)]
    )
    dependencies = explicitly_provided_deps.remaining_after_disambiguation(
        addresses.union(explicitly_provided_deps.includes),
        owners_must_be_ancestors=False,
    )

    return InferredDependencies(dependencies)


# -----------------------------------------------------------------------------------------------
# `openapi_source` dependency inference
# -----------------------------------------------------------------------------------------------


@dataclass(frozen=True)
class OpenApiSourceDependenciesInferenceFieldSet(FieldSet):
    required_fields = (OpenApiSourceField, OpenApiSourceDependenciesField)

    sources: OpenApiSourceField
    dependencies: OpenApiSourceDependenciesField


class InferOpenApiSourceDependenciesRequest(InferDependenciesRequest):
    infer_from = OpenApiSourceDependenciesInferenceFieldSet


@rule
async def infer_openapi_module_dependencies(
    request: InferOpenApiSourceDependenciesRequest,
) -> InferredDependencies:
    explicitly_provided_deps, hydrated_sources = await concurrently(
        determine_explicitly_provided_dependencies(
            **implicitly(DependenciesRequest(request.field_set.dependencies))
        ),
        hydrate_sources(HydrateSourcesRequest(request.field_set.sources), **implicitly()),
    )
    result = await parse_openapi_sources(
        ParseOpenApiSources(
            sources_digest=hydrated_sources.snapshot.digest,
            paths=hydrated_sources.snapshot.files,
        )
    )

    paths: set[str] = set()

    for source_file in hydrated_sources.snapshot.files:
        paths.update(result.dependencies[source_file])

    candidate_targets = await resolve_targets(
        **implicitly(
            RawSpecs(
                file_literals=tuple(FileLiteralSpec(path) for path in paths),
                unmatched_glob_behavior=GlobMatchErrorBehavior.ignore,
                description_of_origin="the `openapi_source` dependency inference",
            )
        )
    )

    addresses = frozenset(
        [target.address for target in candidate_targets if target.has_field(OpenApiSourceField)]
    )
    dependencies = explicitly_provided_deps.remaining_after_disambiguation(
        addresses.union(explicitly_provided_deps.includes),
        owners_must_be_ancestors=False,
    )

    return InferredDependencies(dependencies)


def rules():
    return [
        *collect_rules(),
        UnionRule(InferDependenciesRequest, InferOpenApiDocumentDependenciesRequest),
        UnionRule(InferDependenciesRequest, InferOpenApiSourceDependenciesRequest),
    ]

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	5✔
4
5	import contextlib	5✔
6	import json	5✔
7	import os.path	5✔
8	from collections.abc import Mapping	5✔
9	from dataclasses import dataclass	5✔
10	from typing import Any	5✔
11
12	import yaml	5✔
13
14	from pants.backend.openapi.target_types import (	5✔
15	OPENAPI_FILE_EXTENSIONS,
16	OpenApiDocumentDependenciesField,
17	OpenApiDocumentField,
18	OpenApiSourceDependenciesField,
19	OpenApiSourceField,
20	)
21	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	5✔
22	from pants.base.specs import FileLiteralSpec, RawSpecs	5✔
23	from pants.engine.fs import Digest	5✔
24	from pants.engine.internals.graph import (	5✔
25	determine_explicitly_provided_dependencies,
26	hydrate_sources,
27	resolve_targets,
28	)
29	from pants.engine.internals.selectors import concurrently	5✔
30	from pants.engine.intrinsics import get_digest_contents	5✔
31	from pants.engine.rules import collect_rules, implicitly, rule	5✔
32	from pants.engine.target import (	5✔
33	DependenciesRequest,
34	FieldSet,
35	HydrateSourcesRequest,
36	InferDependenciesRequest,
37	InferredDependencies,
38	)
39	from pants.engine.unions import UnionRule	5✔
40	from pants.util.frozendict import FrozenDict	5✔
41
42
43	@dataclass(frozen=True)	5✔
44	class ParseOpenApiSources:	5✔
45	sources_digest: Digest	5✔
46	paths: tuple[str, ...]	5✔
47
48
49	@dataclass(frozen=True)	5✔
50	class OpenApiDependencies:	5✔
51	dependencies: FrozenDict[str, frozenset[str]]	5✔
52
53
54	@rule	5✔
55	async def parse_openapi_sources(request: ParseOpenApiSources) -> OpenApiDependencies:	5✔
56	digest_contents = await get_digest_contents(request.sources_digest)	4✔
57	dependencies: dict[str, frozenset[str]] = {}	4✔
58
59	for digest_content in digest_contents:	4✔
60	spec = None	4✔
61
62	if digest_content.path.endswith(".json"):	4✔
63	with contextlib.suppress(json.JSONDecodeError):	3✔
64	spec = json.loads(digest_content.content)	3✔
65	elif digest_content.path.endswith(".yaml") or digest_content.path.endswith(".yml"):	3✔
66	with contextlib.suppress(yaml.YAMLError):	3✔
67	spec = yaml.safe_load(digest_content.content)	3✔
68
69	if not spec or not isinstance(spec, dict):	4✔
70	dependencies[digest_content.path] = frozenset()	2✔
71	continue	2✔
72
73	dependencies[digest_content.path] = _find_local_refs(digest_content.path, spec)	4✔
74
75	return OpenApiDependencies(dependencies=FrozenDict(dependencies))	4✔
76
77
78	def _find_local_refs(path: str, d: Mapping[str, Any]) -> frozenset[str]:	5✔
79	local_refs: set[str] = set()	4✔
80
81	for k, v in d.items():	4✔
82	if isinstance(v, dict):	4✔
83	local_refs.update(_find_local_refs(path, v))	3✔
84	elif k == "$ref" and isinstance(v, str):	4✔
85	# https://swagger.io/specification/#reference-object
86	# https://datatracker.ietf.org/doc/html/draft-pbryan-zyp-json-ref-03
87	v = v.split("#", 1)[0]	4✔
88
89	if any(v.endswith(ext) for ext in OPENAPI_FILE_EXTENSIONS) and "://" not in v:	4✔
90	# Resolution is performed relative to the referring document.
91	normalized = os.path.normpath(os.path.join(os.path.dirname(path), v))	4✔
92
93	if not normalized.startswith("../"):	4✔
94	local_refs.add(normalized)	4✔
95
96	return frozenset(local_refs)	4✔
97
98
99	# -----------------------------------------------------------------------------------------------
100	# `openapi_document` dependency inference
101	# -----------------------------------------------------------------------------------------------
102
103
104	@dataclass(frozen=True)	5✔
105	class OpenApiDocumentDependenciesInferenceFieldSet(FieldSet):	5✔
106	required_fields = (OpenApiDocumentField, OpenApiDocumentDependenciesField)	5✔
107
108	sources: OpenApiDocumentField	5✔
109	dependencies: OpenApiDocumentDependenciesField	5✔
110
111
112	class InferOpenApiDocumentDependenciesRequest(InferDependenciesRequest):	5✔
113	infer_from = OpenApiDocumentDependenciesInferenceFieldSet	5✔
114
115
116	@rule	5✔
117	async def infer_openapi_document_dependencies(	5✔
118	request: InferOpenApiDocumentDependenciesRequest,
119	) -> InferredDependencies:
120	explicitly_provided_deps, hydrated_sources = await concurrently(	3✔
121	determine_explicitly_provided_dependencies(
122	**implicitly(DependenciesRequest(request.field_set.dependencies))
123	),
124	hydrate_sources(HydrateSourcesRequest(request.field_set.sources), **implicitly()),
125	)
126	candidate_targets = await resolve_targets(	3✔
127	**implicitly(
128	RawSpecs(
129	file_literals=(FileLiteralSpec(*hydrated_sources.snapshot.files),),
130	description_of_origin="the `openapi_document` dependency inference",
131	)
132	)
133	)
134
135	addresses = frozenset(	3✔
136	[target.address for target in candidate_targets if target.has_field(OpenApiSourceField)]
137	)
138	dependencies = explicitly_provided_deps.remaining_after_disambiguation(	3✔
139	addresses.union(explicitly_provided_deps.includes),
140	owners_must_be_ancestors=False,
141	)
142
143	return InferredDependencies(dependencies)	3✔
144
145
146	# -----------------------------------------------------------------------------------------------
147	# `openapi_source` dependency inference
148	# -----------------------------------------------------------------------------------------------
149
150
151	@dataclass(frozen=True)	5✔
152	class OpenApiSourceDependenciesInferenceFieldSet(FieldSet):	5✔
153	required_fields = (OpenApiSourceField, OpenApiSourceDependenciesField)	5✔
154
155	sources: OpenApiSourceField	5✔
156	dependencies: OpenApiSourceDependenciesField	5✔
157
158
159	class InferOpenApiSourceDependenciesRequest(InferDependenciesRequest):	5✔
160	infer_from = OpenApiSourceDependenciesInferenceFieldSet	5✔
161
162
163	@rule	5✔
164	async def infer_openapi_module_dependencies(	5✔
165	request: InferOpenApiSourceDependenciesRequest,
166	) -> InferredDependencies:
167	explicitly_provided_deps, hydrated_sources = await concurrently(	3✔
168	determine_explicitly_provided_dependencies(
169	**implicitly(DependenciesRequest(request.field_set.dependencies))
170	),
171	hydrate_sources(HydrateSourcesRequest(request.field_set.sources), **implicitly()),
172	)
173	result = await parse_openapi_sources(	3✔
174	ParseOpenApiSources(
175	sources_digest=hydrated_sources.snapshot.digest,
176	paths=hydrated_sources.snapshot.files,
177	)
178	)
179
180	paths: set[str] = set()	3✔
181
182	for source_file in hydrated_sources.snapshot.files:	3✔
183	paths.update(result.dependencies[source_file])	3✔
184
185	candidate_targets = await resolve_targets(	3✔
186	**implicitly(
187	RawSpecs(
188	file_literals=tuple(FileLiteralSpec(path) for path in paths),
189	unmatched_glob_behavior=GlobMatchErrorBehavior.ignore,
190	description_of_origin="the `openapi_source` dependency inference",
191	)
192	)
193	)
194
195	addresses = frozenset(	3✔
196	[target.address for target in candidate_targets if target.has_field(OpenApiSourceField)]
197	)
198	dependencies = explicitly_provided_deps.remaining_after_disambiguation(	3✔
199	addresses.union(explicitly_provided_deps.includes),
200	owners_must_be_ancestors=False,
201	)
202
203	return InferredDependencies(dependencies)	3✔
204
205
206	def rules():	5✔
207	return [	5✔
208	*collect_rules(),
209	UnionRule(InferDependenciesRequest, InferOpenApiDocumentDependenciesRequest),
210	UnionRule(InferDependenciesRequest, InferOpenApiSourceDependenciesRequest),
211	]

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