Method lock

pex/resolve/lockfile/create.py:334–483

(self, downloaded)

        )

    def lock(self, downloaded):
        # type: (Downloaded) -> Tuple[LockedResolve, ...]

        dist_metadatas_by_download_target = defaultdict(
            OrderedSet
        )  # type: DefaultDict[DownloadTarget, OrderedSet[DistMetadata]]
        build_requests = OrderedSet()  # type: OrderedSet[BuildRequest]

        for local_distribution in downloaded.local_distributions:
            if local_distribution.is_wheel:
                dist_metadatas_by_download_target[local_distribution.download_target].add(
                    DistMetadata.load(local_distribution.path)
                )
            elif os.path.isfile(local_distribution.path):
                build_requests.add(
                    BuildRequest.for_file(
                        target=local_distribution.download_target,
                        source_path=local_distribution.path,
                        subdirectory=local_distribution.subdirectory,
                    )
                )
            else:
                build_requests.add(
                    BuildRequest.for_directory(
                        target=local_distribution.download_target,
                        source_path=local_distribution.path,
                        subdirectory=local_distribution.subdirectory,
                    )
                )

        resolved_requirements_by_download_target = (
            OrderedDict()
        )  # type: OrderedDict[DownloadTarget, Tuple[ResolvedRequirement, ...]]
        for analysis in self._analysis:
            lock_result = analysis.analyzer.lock_result
            build_requests.update(
                BuildRequest.for_directory(
                    target=analysis.download_target, source_path=local_project
                )
                for local_project in lock_result.local_projects
            )
            resolved_requirements_by_download_target[
                analysis.download_target
            ] = lock_result.resolved_requirements

        with TRACER.timed(
            "Building {count} source {distributions} to gather metadata for lock.".format(
                count=len(build_requests), distributions=pluralize(build_requests, "distribution")
            )
        ):
            pool = ThreadPool(processes=self.max_parallel_jobs)
            try:
                targets_and_project_directories = list(
                    pool.map(_prepare_project_directory, build_requests),
                )
            finally:
                pool.close()
                pool.join()