forked from pantsbuild/pants
-
Notifications
You must be signed in to change notification settings - Fork 0
/
dependencies.py
59 lines (49 loc) · 2.21 KB
/
dependencies.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from pants.backend.docker.subsystems.dockerfile_parser import DockerfileInfo, DockerfileInfoRequest
from pants.backend.docker.target_types import DockerDependenciesField
from pants.base.specs import AddressSpecs, MaybeEmptySiblingAddresses
from pants.core.goals.package import PackageFieldSet
from pants.engine.addresses import Addresses, UnparsedAddressInputs
from pants.engine.rules import Get, collect_rules, rule
from pants.engine.target import (
FieldSetsPerTarget,
FieldSetsPerTargetRequest,
InjectDependenciesRequest,
InjectedDependencies,
Targets,
)
from pants.engine.unions import UnionRule
class InjectDockerDependencies(InjectDependenciesRequest):
inject_for = DockerDependenciesField
@rule
async def inject_docker_dependencies(request: InjectDockerDependencies) -> InjectedDependencies:
"""Inspects COPY instructions in the Dockerfile for references to known packagable targets."""
dockerfile_info = await Get(
DockerfileInfo, DockerfileInfoRequest(request.dependencies_field.address)
)
# Parse all putative target addresses.
putative_addresses = await Get(
Addresses,
UnparsedAddressInputs(
dockerfile_info.putative_target_addresses,
owning_address=dockerfile_info.address,
),
)
# Get the target for those addresses that are known.
directories = {address.spec_path for address in putative_addresses}
all_addresses = await Get(Addresses, AddressSpecs(map(MaybeEmptySiblingAddresses, directories)))
targets = await Get(
Targets, Addresses((address for address in putative_addresses if address in all_addresses))
)
# Only keep those targets that we can "package".
package = await Get(FieldSetsPerTarget, FieldSetsPerTargetRequest(PackageFieldSet, targets))
referenced_targets = (
field_sets[0].address for field_sets in package.collection if len(field_sets) > 0
)
return InjectedDependencies(Addresses(referenced_targets))
def rules():
return [
*collect_rules(),
UnionRule(InjectDependenciesRequest, InjectDockerDependencies),
]