19000741080

Committed 01 Nov 2025 06:16PM UTC coverage: 80.3% (+0.3%) from 80.004%

Build # 19000741080

Build Type

Pull #22837

github

Committed by

web-flow

Commit Message

Merge 51f49bc90 into da3fb359e

Pull Request Pull Request #22837: Updated Treesitter dependencies

Run Details

77994 of 97128 relevant lines covered (80.3%)

3.35 hits per line

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

58.06

/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(  # 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	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✔
37	return FrozenDict((label, app.name) for label, app in self.items())	×
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())	×
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"	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())	×
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_configured_default(
108	python_setup, tgt[PythonResolveField] if tgt.has_field(PythonResolveField) else None
109	)
110	)
111	ics_to_tgts[ics].append(tgt)	×
112
113	for ics, tgts in ics_to_tgts.items():	×
114	sources = await concurrently( # noqa: PNT30: requires triage	×
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]))	×
121	prefixed_apps_digest = await add_prefix(AddPrefix(apps_digest, apps_sandbox_prefix))	×
122
123	input_digest = await merge_digests(MergeDigests([prefixed_apps_digest, script_digest]))	×
124	python_interpreter = await find_interpreter(ics, **implicitly())	×
125
126	process_result = await execute_process_or_raise(	×
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(	×
140	process_result.stdout or b"{}", strip_prefix=apps_sandbox_prefix
141	)
142
143	return django_apps	×
144
145
146	def rules() -> Iterable[Rule]:	2✔
147	return collect_rules()	2✔

pantsbuild / pants / 19000741080

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