pantsbuild / pants / 19015773527

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).

# Licensed under the Apache License, Version 2.0 (see LICENSE).

    DEFAULT_MODULE_MAPPING,

    DEFAULT_MODULE_PATTERN_MAPPING,

    DEFAULT_TYPE_STUB_MODULE_MAPPING,

    DEFAULT_TYPE_STUB_MODULE_PATTERN_MAPPING,

)

    PythonRequirementModulesField,

    PythonRequirementResolveField,

    PythonRequirementsField,

    PythonRequirementTypeStubModulesField,

    PythonResolveField,

    PythonSourceField,

)

    # 0 = The provider mapped to the module itself, 1 = the provider mapped to its parent, etc.

        path.parent if path.name in ("__init__.py", "__init__.pyi") else path.with_suffix("")

    )

# -----------------------------------------------------------------------------------------------

# First-party module mapping

# -----------------------------------------------------------------------------------------------

    FrozenDict[ResolveName, FrozenDict[str, tuple[ModuleProvider, ...]]]

):

    """A mapping of each resolve name to the first-party module names contained and their owning

    addresses.

    This contains the modules from a specific implementation, e.g. a codegen backend. All

    implementations then get merged.

    """

        cls,

        resolves_to_modules_to_providers: Mapping[

            ResolveName, Mapping[str, Iterable[ModuleProvider]]

        ],

    ) -> FirstPartyPythonMappingImpl:

            (

                resolve,

                FrozenDict(

                    (mod, tuple(sorted(providers))) for mod, providers in sorted(mapping.items())

                ),

            )

            for resolve, mapping in sorted(resolves_to_modules_to_providers.items())

        )

    """An entry point for a specific implementation of mapping module names to owning targets for

    Python import dependency inference.

    All implementations will be merged together. Any modules that show up in multiple

    implementations will be marked ambiguous.

    """

    marker: FirstPartyPythonMappingImplMarker, env_name: EnvironmentName

) -> FirstPartyPythonMappingImpl:

        ResolveName, FrozenDict[str, tuple[ModuleProvider, ...]]

    ]

    owning addresses.

    This mapping may have been constructed from multiple distinct implementations, e.g.

    implementations for each codegen backends.

    """

        self, module: str, resolve: str

    ) -> tuple[PossibleModuleProvider, ...]:

        # If the module is not found, try the parent, if any. This is to handle `from` imports

        # where the "module" we were handed was actually a symbol inside the module.

        # E.g., with `from my_project.app import App`, we would be passed "my_project.app.App".

        #

        # We do not look past the direct parent, as this could cause multiple ambiguous owners to

        # be resolved. This contrasts with the third-party module mapping, which will try every

        # ancestor.

        # TODO: Now that we capture the ancestry, we could look past the direct parent.

        #  One reason to do so would be to unify more of the FirstParty and ThirdParty impls.

        self, module: str, resolve: str | None

    ) -> tuple[PossibleModuleProvider, ...]:

        """Find all providers for the module.

        If `resolve` is None, will not consider resolves, i.e. any `python_source` et al can be

        used. Otherwise, providers can only come from first-party targets with the resolve.

        """

            itertools.chain.from_iterable(

                self._providers_for_resolve(module, resolve)

                for resolve in list(self.resolves_to_modules_to_providers.keys())

            )

        )

    union_membership: UnionMembership,

) -> FirstPartyPythonModuleMapping:

        get_first_party_python_mapping_impl(

            **implicitly({marker_cls(): FirstPartyPythonMappingImplMarker})

        )

        for marker_cls in union_membership.get(FirstPartyPythonMappingImplMarker)

    )

        ResolveName, DefaultDict[str, list[ModuleProvider]]

    ] = defaultdict(lambda: defaultdict(list))

        FrozenDict(

            (

                resolve,

                FrozenDict(

                    (mod, tuple(sorted(providers))) for mod, providers in sorted(mapping.items())

                ),

            )

            for resolve, mapping in sorted(resolves_to_modules_to_providers.items())

        )

    )

# This is only used to register our implementation with the plugin hook via unions. Note that we

# implement this like any other plugin implementation so that we can run them all in parallel.

    desc="Creating map of first party Python targets to Python modules",

    level=LogLevel.DEBUG,

)

    _: FirstPartyPythonTargetsMappingMarker,

    all_python_targets: AllPythonTargets,

    python_setup: PythonSetup,

) -> FirstPartyPythonMappingImpl:

        strip_file_name(StrippedFileNameRequest(tgt[PythonSourceField].file_path))

        for tgt in all_python_targets.first_party

    )

        ResolveName, DefaultDict[str, list[ModuleProvider]]

    ] = defaultdict(lambda: defaultdict(list))

            ModuleProviderType.TYPE_STUB if stripped_f.suffix == ".pyi" else ModuleProviderType.IMPL

        )

            ModuleProvider(tgt.address, provider_type)

        )

# -----------------------------------------------------------------------------------------------

# Third party module mapping

# -----------------------------------------------------------------------------------------------

    """A mapping of each resolve to the modules they contain and the addresses providing those

    modules."""

        ResolveName, FrozenDict[str, tuple[ModuleProvider, ...]]

    ]

        self, module: str, resolve: str, ancestry: int = 0

    ) -> tuple[PossibleModuleProvider, ...]:

        # If the module is not found, recursively try the ancestor modules, if any. For example,

        # pants.task.task.Task -> pants.task.task -> pants.task -> pants

        self, module: str, resolve: str | None

    ) -> tuple[PossibleModuleProvider, ...]:

        """Find all providers for the module.

        If `resolve` is None, will not consider resolves, i.e. any `python_requirement` can be

        consumed. Otherwise, providers can only come from `python_requirements` with the resolve.

        """

            itertools.chain.from_iterable(

                self._providers_for_resolve(module, resolve)

                for resolve in list(self.resolves_to_modules_to_providers.keys())

            )

        )

    """Generate a tuple of possible module mappings from a project name using a regex pattern.

    e.g. google-cloud-foo -> [google.cloud.foo, google.cloud.foo_v1, google.cloud.foo_v2]

    Should eliminate the need to "manually" add a mapping for every service

    proj_name: The project name to generate mappings for e.g google-cloud-datastream

    """

        DEFAULT_TYPE_STUB_MODULE_PATTERN_MAPPING if is_type_stub else DEFAULT_MODULE_PATTERN_MAPPING

    )

                match_pattern.sub(replace_pattern, proj_name)

                for replace_pattern in replace_patterns

            ]

    all_python_targets: AllPythonTargets,

    python_setup: PythonSetup,

) -> ThirdPartyPythonModuleMapping:

        ResolveName, DefaultDict[str, list[ModuleProvider]]

    ] = defaultdict(lambda: defaultdict(list))

                    ModuleProvider(

                        tgt.address,

                        ModuleProviderType.TYPE_STUB if is_type_stub else ModuleProviderType.IMPL,

                    )

                )

        # Else, fall back to defaults.

            # NB: We don't use `canonicalize_project_name()` for the fallback value because we

            # want to preserve `.` in the module name. See

            # https://www.python.org/dev/peps/pep-0503/#normalized-names.

            modules_to_add: tuple[str, ...]

            is_type_stub: bool

            # check for stubs first, since stub packages may also match impl package patterns

            else:

        FrozenDict(

            (

                resolve,

                FrozenDict(

                    (mod, tuple(sorted(providers))) for mod, providers in sorted(mapping.items())

                ),

            )

            for resolve, mapping in sorted(resolves_to_modules_to_providers.items())

        )

    )

# -----------------------------------------------------------------------------------------------

# module -> owners

# -----------------------------------------------------------------------------------------------

    """The target(s) that own a Python module.

    Up to 2 targets can unambiguously own the same module, if one is an implementation and the other

    is a .pyi type stub. It is ambiguous for >1 implementation target to own the same module, and

    those targets will be put into `ambiguous` instead of `unambiguous`. Therefore, `unambiguous`

    should never be >2; and only 1 of `unambiguous` and `ambiguous` should have targets.

    """

                softwrap(

                    """

                    A module has both unambiguous and ambiguous owners, which is a bug in the

                    dependency inference code. Please file a bug report at

                    https://github.com/pantsbuild/pants/issues/new.

                    """

                )

            )

    # If specified, resolve ambiguity by choosing the symbol provider with the

    # closest common ancestor to this path. Must be a path relative to the build root.

    request: PythonModuleOwnersRequest,

    first_party_mapping: FirstPartyPythonModuleMapping,

    third_party_mapping: ThirdPartyPythonModuleMapping,

) -> PythonModuleOwners:

        *third_party_mapping.providers_for_module(request.module, resolve=request.resolve),

        *first_party_mapping.providers_for_module(request.module, resolve=request.resolve),

    )

    # We first attempt to disambiguate conflicting providers by taking - for each provider type -

    # the providers of the closest ancestors to the requested modules.

    # E.g., if we have a provider for foo.bar and for foo.bar.baz, prefer the latter.

    # This prevents issues with namespace packages that are split between first-party and

    # third-party (e.g., https://github.com/pantsbuild/pants/discussions/17286).

    # Map from provider type to mutable pair of

    # [closest ancestry, list of provider of that type at that ancestry level].

        # NB This must come after the < check above, so we handle the possible_provider

        # that caused that check to pass.

        # For each provider type, if we have more than one provider left, prefer

        # the one with the closest common ancestor to the requester.

                    os.path.commonpath([request.locality, provider.addr.spec_path])

                )

        itertools.chain(*[val[1] for val in type_to_closest_providers.values()])

    )

    # Check that we have at most one remaining provider for each provider type.

    # If we have more than one, signal ambiguity.

        *collect_rules(),

        UnionRule(FirstPartyPythonMappingImplMarker, FirstPartyPythonTargetsMappingMarker),

    )