Plugin with custom target to work on python source files

README.md

Example output (edited for readability):

$ ./pants package src/example:dags-demo
17:16:25.12 [INFO] Initializing scheduler...
17:16:25.41 [INFO] Scheduler initialized.
17:16:27.41 [INFO] (

HydratedSources(
  snapshot=Snapshot(
    digest=(89d98482b939d9f4b99c1ad4bfcf8ba49344fd536459773f4e70ac019f5914b6, 77), 
    dirs=(src,src/example), 
    files=(src/example/__init__.py)), 
  filespec={'includes': ['src/example/__init__.py']}, 
  sources_type=<class 'pants.backend.python.target_types.PythonSources'>), 

HydratedSources(
  snapshot=Snapshot(
    digest=(970b0af53d2b5a86e9070f95b202660e902d02fb9fdceed0b223a86dfb9d79b2, 77), 
    dirs=(src,src/example), 
    files=(src/example/demo.py)), 
  filespec={'includes': ['src/example/demo.py']}, 
  sources_type=<class 'pants.backend.python.target_types.PythonSources'>))

17:16:27.41 [INFO] Completed: dags_plugin.register.package_dags

Directory structure:

.
├── pants
├── pants-plugins
│   └── dags_plugin
│       ├── BUILD
│       ├── __init__.py
│       └── register.py
├── pants.toml
└── src
    └── example
        ├── BUILD
        ├── __init__.py
        └── demo.py

4 directories, 8 files

pants-plugins dags_plugin BUILD

python_library()

pants-plugins dags_plugin register.py

from dataclasses import dataclass

from pants.backend.python.target_types import PythonSources
from pants.core.goals.package import (BuiltPackage, OutputPathField,
                                      PackageFieldSet)
from pants.engine.fs import EMPTY_DIGEST
from pants.engine.rules import Get, MultiGet, collect_rules, rule
from pants.engine.target import (COMMON_TARGET_FIELDS, Dependencies,
                                 HydratedSources, HydrateSourcesRequest,
                                 Sources, Target, TransitiveTargets,
                                 TransitiveTargetsRequest)
from pants.engine.unions import UnionRule
from pants.util.logging import LogLevel
import logging


class DagsDependencies(Dependencies):
    """Simple inheritance for dependencies."""


class Dags(Target):
    """DAG tasks generated by a library."""

    alias = "dags"
    core_fields = (*COMMON_TARGET_FIELDS, DagsDependencies)
    help = "Dags target"


@dataclass(frozen=True)
class DagsFieldSet(PackageFieldSet):
    required_fields = (DagsDependencies,)

    output_path: OutputPathField
    dependencies: DagsDependencies


@rule(level=LogLevel.INFO)
async def package_dags(field_set: DagsFieldSet) -> BuiltPackage:
    output_filename = field_set.output_path.value_or_default(
        field_set.address, file_ending="yaml"
    )

    transitive_targets = await Get(
        TransitiveTargets,
        TransitiveTargetsRequest([field_set.address]),
    )

    sources = await MultiGet(
        Get(
            HydratedSources,
            HydrateSourcesRequest(
                tgt.get(Sources),
                for_sources_types=(PythonSources,),
                # enable_codegen=True,  # optional
            ),
        )
        for tgt in transitive_targets.dependencies
    )

    # Do something with sources.
    logging.info(str(sources))

    # Return resulting digest.. ?
    return BuiltPackage(
        EMPTY_DIGEST,
        tuple(
            # BuiltPackageArtifact()
        ),
    )


def target_types():
    return [Dags]


def rules():
    return [
        *collect_rules(),
        UnionRule(PackageFieldSet, DagsFieldSet),
    ]

pants.toml

[GLOBAL]
pants_version = "2.7.0"

pythonpath = ["%(buildroot)s/pants-plugins"]
backend_packages.add = [
  "pants.backend.python",
  "pants.backend.python.lint.black",
  "pants.backend.python.lint.isort",
  "dags_plugin",
]

[anonymous-telemetry]
enabled = false

src example BUILD

python_library(name="example")

dags(
    name="dags-demo",
    dependencies=[
        ":example",
    ],
)

src example demo.py

def fun():
    return "Hello World"