25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

1294 existing lines in 76 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

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

96.77

/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, PythonResolveField
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_configured_default(
                python_setup, tgt[PythonResolveField] if tgt.has_field(PythonResolveField) else None
            )
        )
        ics_to_tgts[ics].append(tgt)

    for ics, tgts in ics_to_tgts.items():
        sources = await concurrently(
            [
                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	2✔
4
5	import json	2✔
6	import os	2✔
7	from collections import defaultdict	2✔
8	from collections.abc import Iterable	2✔
9	from dataclasses import dataclass	2✔
10
11	from pants.backend.python.subsystems.setup import PythonSetup	2✔
12	from pants.backend.python.target_types import InterpreterConstraintsField, PythonResolveField	2✔
13	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	2✔
14	from pants.backend.python.util_rules.pex import find_interpreter	2✔
15	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	2✔
16	from pants.base.specs import FileGlobSpec, RawSpecs	2✔
17	from pants.engine.fs import AddPrefix, CreateDigest, FileContent, MergeDigests	2✔
18	from pants.engine.internals.graph import hydrate_sources, resolve_targets	2✔
19	from pants.engine.internals.selectors import concurrently	2✔
20	from pants.engine.intrinsics import add_prefix, create_digest, merge_digests	2✔
21	from pants.engine.process import Process, execute_process_or_raise	2✔
22	from pants.engine.rules import Rule, collect_rules, implicitly, rule	2✔
23	from pants.engine.target import HydrateSourcesRequest, SourcesField, Target	2✔
24	from pants.util.frozendict import FrozenDict	2✔
25	from pants.util.resources import read_resource	2✔
26
27
28	@dataclass(frozen=True)	2✔
29	class DjangoApp:	2✔
30	name: str	2✔
31	config_file: str	2✔
32
33
34	class DjangoApps(FrozenDict[str, DjangoApp]):	2✔
35	@property	2✔
36	def label_to_name(self) -> FrozenDict[str, str]:	2✔
UNCOV 37	return FrozenDict((label, app.name) for label, app in self.items())	1✔
38
39	@property	2✔
40	def label_to_file(self) -> FrozenDict[str, str]:	2✔
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:	2✔
44	json_dict: dict[str, dict[str, str]] = json.loads(json_bytes.decode())	2✔
45	apps = {	2✔
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)	2✔
52	return DjangoApps(sorted(combined.items()))	2✔
53
54
55	_script_resource = "scripts/app_detector.py"	2✔
56
57
58	@rule	2✔
59	async def detect_django_apps(python_setup: PythonSetup) -> DjangoApps:	2✔
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())	2✔
85	targets = await resolve_targets(	2✔
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:	2✔
95	return django_apps	×
96
97	script_file_content = FileContent(	2✔
98	"script/__visitor.py", read_resource(__name__, _script_resource)
99	)
100	script_digest = await create_digest(CreateDigest([script_file_content]))	2✔
101	apps_sandbox_prefix = "_apps_to_detect"	2✔
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)	2✔
105	for tgt in targets:	2✔
106	ics = InterpreterConstraints(	2✔
107	tgt[InterpreterConstraintsField].value_or_configured_default(
108	python_setup, tgt[PythonResolveField] if tgt.has_field(PythonResolveField) else None
109	)
110	)
111	ics_to_tgts[ics].append(tgt)	2✔
112
113	for ics, tgts in ics_to_tgts.items():	2✔
114	sources = await concurrently(	2✔
115	[
116	hydrate_sources(HydrateSourcesRequest(tgt[SourcesField]), **implicitly())
117	for tgt in tgts
118	]
119	)
120	apps_digest = await merge_digests(MergeDigests([src.snapshot.digest for src in sources]))	2✔
121	prefixed_apps_digest = await add_prefix(AddPrefix(apps_digest, apps_sandbox_prefix))	2✔
122
123	input_digest = await merge_digests(MergeDigests([prefixed_apps_digest, script_digest]))	2✔
124	python_interpreter = await find_interpreter(ics, **implicitly())	2✔
125
126	process_result = await execute_process_or_raise(	2✔
127	**implicitly(
128	Process(
129	argv=[
130	python_interpreter.path,
131	script_file_content.path,
132	apps_sandbox_prefix,
133	],
134	input_digest=input_digest,
135	description="Detect Django apps",
136	)
137	)
138	)
139	django_apps = django_apps.add_from_json(	2✔
140	process_result.stdout or b"{}", strip_prefix=apps_sandbox_prefix
141	)
142
143	return django_apps	2✔
144
145
146	def rules() -> Iterable[Rule]:	2✔
147	return collect_rules()	2✔

pantsbuild / pants / 25443604553

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