18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

0.0

/src/python/pants/backend/python/framework/django/detect_apps.py

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

import json
import os
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass

from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import InterpreterConstraintsField
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.pex import find_interpreter
from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.base.specs import FileGlobSpec, RawSpecs
from pants.engine.fs import AddPrefix, CreateDigest, FileContent, MergeDigests
from pants.engine.internals.graph import hydrate_sources, resolve_targets
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import add_prefix, create_digest, merge_digests
from pants.engine.process import Process, execute_process_or_raise
from pants.engine.rules import Rule, collect_rules, implicitly, rule
from pants.engine.target import HydrateSourcesRequest, SourcesField, Target
from pants.util.frozendict import FrozenDict
from pants.util.resources import read_resource


@dataclass(frozen=True)
class DjangoApp:
    name: str
    config_file: str


class DjangoApps(FrozenDict[str, DjangoApp]):
    @property
    def label_to_name(self) -> FrozenDict[str, str]:
        return FrozenDict((label, app.name) for label, app in self.items())

    @property
    def label_to_file(self) -> FrozenDict[str, str]:
        return FrozenDict((label, app.config_file) for label, app in self.items())

    def add_from_json(self, json_bytes: bytes, strip_prefix: str = "") -> DjangoApps:
        json_dict: dict[str, dict[str, str]] = json.loads(json_bytes.decode())
        apps = {
            label: DjangoApp(
                val["app_name"], val["config_file"].partition(f"{strip_prefix}{os.sep}")[2]
            )
            for label, val in json_dict.items()
        }
        combined = dict(self, **apps)
        return DjangoApps(sorted(combined.items()))


_script_resource = "scripts/app_detector.py"


@rule
async def detect_django_apps(python_setup: PythonSetup) -> DjangoApps:
    # A Django app has a "name" - the full import path to the app ("path.to.myapp"),
    # and a "label" - a short name, usually the last segment of the import path ("myapp").
    #
    # An app provides this information via a subclass of AppConfig, living in a
    # file named apps.py.  Django loads this information into an app registry at runtime.
    #
    # Some parts of Django, notably migrations, use the label to reference apps. So to do custom
    # Django dep inference, we need to know the label -> name mapping.
    #
    # The only truly correct way to enumerate Django apps is to run the Django app registry code.
    # However we can't do this until after dep inference has completed, and even then it would be
    # complicated: we wouldn't know which settings.py to use, or whether it's safe to run Django
    # against that settings.py. Instead, we do this statically via parsing the apps.py file.
    #
    # NB: Legacy Django apps may not have an apps.py, in which case the label is assumed to be
    #  the name of the app dir, but the recommendation for many years has been to have it, and
    #  the Django startapp tool creates it for you. If an app does not have such an apps.py,
    #  then we won't be able to infer deps on that app unless we find other ways of detecting it.
    #  We should only do that if that case turns out to be common, and for some reason users can't
    #  simply create an apps.py to fix the issue.
    #
    # NB: Right now we only detect first-party apps in repo. We assume that third-party apps will
    #  be dep-inferred as a whole via the full package path in settings.py anyway.
    #  In the future we may find a way to map third-party apps here as well.
    django_apps = DjangoApps(FrozenDict())
    targets = await resolve_targets(
        **implicitly(
            RawSpecs.create(
                specs=[FileGlobSpec("**/apps.py")],
                description_of_origin="Django app detection",
                unmatched_glob_behavior=GlobMatchErrorBehavior.ignore,
            )
        )
    )
    if not targets:
        return django_apps

    script_file_content = FileContent(
        "script/__visitor.py", read_resource(__name__, _script_resource)
    )
    script_digest = await create_digest(CreateDigest([script_file_content]))
    apps_sandbox_prefix = "_apps_to_detect"

    # Partition by ICs, so we can run the detector on the appropriate interpreter.
    ics_to_tgts: dict[InterpreterConstraints, list[Target]] = defaultdict(list)
    for tgt in targets:
        ics = InterpreterConstraints(
            tgt[InterpreterConstraintsField].value_or_global_default(python_setup)
        )
        ics_to_tgts[ics].append(tgt)

    for ics, tgts in ics_to_tgts.items():
        sources = await concurrently(  # noqa: PNT30: requires triage
            [
                hydrate_sources(HydrateSourcesRequest(tgt[SourcesField]), **implicitly())
                for tgt in tgts
            ]
        )
        apps_digest = await merge_digests(MergeDigests([src.snapshot.digest for src in sources]))
        prefixed_apps_digest = await add_prefix(AddPrefix(apps_digest, apps_sandbox_prefix))

        input_digest = await merge_digests(MergeDigests([prefixed_apps_digest, script_digest]))
        python_interpreter = await find_interpreter(ics, **implicitly())

        process_result = await execute_process_or_raise(
            **implicitly(
                Process(
                    argv=[
                        python_interpreter.path,
                        script_file_content.path,
                        apps_sandbox_prefix,
                    ],
                    input_digest=input_digest,
                    description="Detect Django apps",
                )
            )
        )
        django_apps = django_apps.add_from_json(
            process_result.stdout or b"{}", strip_prefix=apps_sandbox_prefix
        )

    return django_apps


def rules() -> Iterable[Rule]:
    return collect_rules()

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	×
4
5	import json	×
6	import os	×
7	from collections import defaultdict	×
8	from collections.abc import Iterable	×
9	from dataclasses import dataclass	×
10
11	from pants.backend.python.subsystems.setup import PythonSetup	×
12	from pants.backend.python.target_types import InterpreterConstraintsField	×
13	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	×
14	from pants.backend.python.util_rules.pex import find_interpreter	×
15	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	×
16	from pants.base.specs import FileGlobSpec, RawSpecs	×
17	from pants.engine.fs import AddPrefix, CreateDigest, FileContent, MergeDigests	×
18	from pants.engine.internals.graph import hydrate_sources, resolve_targets	×
19	from pants.engine.internals.selectors import concurrently	×
20	from pants.engine.intrinsics import add_prefix, create_digest, merge_digests	×
21	from pants.engine.process import Process, execute_process_or_raise	×
22	from pants.engine.rules import Rule, collect_rules, implicitly, rule	×
23	from pants.engine.target import HydrateSourcesRequest, SourcesField, Target	×
24	from pants.util.frozendict import FrozenDict	×
25	from pants.util.resources import read_resource	×
26
27
28	@dataclass(frozen=True)	×
29	class DjangoApp:	×
30	name: str	×
31	config_file: str	×
32
33
34	class DjangoApps(FrozenDict[str, DjangoApp]):	×
35	@property	×
36	def label_to_name(self) -> FrozenDict[str, str]:	×
37	return FrozenDict((label, app.name) for label, app in self.items())	×
38
39	@property	×
40	def label_to_file(self) -> FrozenDict[str, str]:	×
41	return FrozenDict((label, app.config_file) for label, app in self.items())	×
42
43	def add_from_json(self, json_bytes: bytes, strip_prefix: str = "") -> DjangoApps:	×
44	json_dict: dict[str, dict[str, str]] = json.loads(json_bytes.decode())	×
45	apps = {	×
46	label: DjangoApp(
47	val["app_name"], val["config_file"].partition(f"{strip_prefix}{os.sep}")[2]
48	)
49	for label, val in json_dict.items()
50	}
51	combined = dict(self, **apps)	×
52	return DjangoApps(sorted(combined.items()))	×
53
54
55	_script_resource = "scripts/app_detector.py"	×
56
57
58	@rule	×
59	async def detect_django_apps(python_setup: PythonSetup) -> DjangoApps:	×
60	# A Django app has a "name" - the full import path to the app ("path.to.myapp"),
61	# and a "label" - a short name, usually the last segment of the import path ("myapp").
62	#
63	# An app provides this information via a subclass of AppConfig, living in a
64	# file named apps.py. Django loads this information into an app registry at runtime.
65	#
66	# Some parts of Django, notably migrations, use the label to reference apps. So to do custom
67	# Django dep inference, we need to know the label -> name mapping.
68	#
69	# The only truly correct way to enumerate Django apps is to run the Django app registry code.
70	# However we can't do this until after dep inference has completed, and even then it would be
71	# complicated: we wouldn't know which settings.py to use, or whether it's safe to run Django
72	# against that settings.py. Instead, we do this statically via parsing the apps.py file.
73	#
74	# NB: Legacy Django apps may not have an apps.py, in which case the label is assumed to be
75	# the name of the app dir, but the recommendation for many years has been to have it, and
76	# the Django startapp tool creates it for you. If an app does not have such an apps.py,
77	# then we won't be able to infer deps on that app unless we find other ways of detecting it.
78	# We should only do that if that case turns out to be common, and for some reason users can't
79	# simply create an apps.py to fix the issue.
80	#
81	# NB: Right now we only detect first-party apps in repo. We assume that third-party apps will
82	# be dep-inferred as a whole via the full package path in settings.py anyway.
83	# In the future we may find a way to map third-party apps here as well.
84	django_apps = DjangoApps(FrozenDict())	×
85	targets = await resolve_targets(	×
86	**implicitly(
87	RawSpecs.create(
88	specs=[FileGlobSpec("**/apps.py")],
89	description_of_origin="Django app detection",
90	unmatched_glob_behavior=GlobMatchErrorBehavior.ignore,
91	)
92	)
93	)
94	if not targets:	×
95	return django_apps	×
96
97	script_file_content = FileContent(	×
98	"script/__visitor.py", read_resource(__name__, _script_resource)
99	)
100	script_digest = await create_digest(CreateDigest([script_file_content]))	×
101	apps_sandbox_prefix = "_apps_to_detect"	×
102
103	# Partition by ICs, so we can run the detector on the appropriate interpreter.
104	ics_to_tgts: dict[InterpreterConstraints, list[Target]] = defaultdict(list)	×
105	for tgt in targets:	×
106	ics = InterpreterConstraints(	×
107	tgt[InterpreterConstraintsField].value_or_global_default(python_setup)
108	)
109	ics_to_tgts[ics].append(tgt)	×
110
111	for ics, tgts in ics_to_tgts.items():	×
112	sources = await concurrently( # noqa: PNT30: requires triage	×
113	[
114	hydrate_sources(HydrateSourcesRequest(tgt[SourcesField]), **implicitly())
115	for tgt in tgts
116	]
117	)
118	apps_digest = await merge_digests(MergeDigests([src.snapshot.digest for src in sources]))	×
119	prefixed_apps_digest = await add_prefix(AddPrefix(apps_digest, apps_sandbox_prefix))	×
120
121	input_digest = await merge_digests(MergeDigests([prefixed_apps_digest, script_digest]))	×
122	python_interpreter = await find_interpreter(ics, **implicitly())	×
123
124	process_result = await execute_process_or_raise(	×
125	**implicitly(
126	Process(
127	argv=[
128	python_interpreter.path,
129	script_file_content.path,
130	apps_sandbox_prefix,
131	],
132	input_digest=input_digest,
133	description="Detect Django apps",
134	)
135	)
136	)
137	django_apps = django_apps.add_from_json(	×
138	process_result.stdout or b"{}", strip_prefix=apps_sandbox_prefix
139	)
140
141	return django_apps	×
142
143
144	def rules() -> Iterable[Rule]:	×
145	return collect_rules()	×

pantsbuild / pants / 18252174847

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