Pants django-run goal

register.py

"""Plugin for django management commands"""
import glob
from pathlib import PurePath
from os.path import join, relpath, exists
from shutil import copyfile

from pants.base.build_root import BuildRoot
from pants.core.goals.run import RunFieldSet, RunRequest, RunSubsystem
from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest
from pants.engine.console import Console
from pants.engine.environment import CompleteEnvironment
from pants.engine.fs import Workspace
from pants.engine.goal import Goal
from pants.engine.process import InteractiveProcess, InteractiveRunner
from pants.engine.rules import Get, collect_rules, goal_rule
from pants.engine.target import (
    FieldSet,
    NoApplicableTargetsBehavior,
    TargetRootsToFieldSets,
    TargetRootsToFieldSetsRequest,
)
from pants.option.global_options import GlobalOptions
from pants.util.contextutil import temporary_dir

from .target_types import DjangoAppTarget, MutableFiles

class DjangoRunSubsystem(RunSubsystem):
    name = "django-run"
    help = "Run Django management commands"


class DjangoRun(Goal):
    subsystem_cls = DjangoRunSubsystem


@goal_rule
async def django_run(
    django_run_subsystem: DjangoRunSubsystem,
    global_options: GlobalOptions,
    console: Console,
    interactive_runner: InteractiveRunner,
    workspace: Workspace,
    build_root: BuildRoot,
    complete_env: CompleteEnvironment,
) -> DjangoRun:
    targets_to_valid_field_sets = await Get(
        TargetRootsToFieldSets,
        TargetRootsToFieldSetsRequest(
            RunFieldSet,
            goal_description="the `run` goal",
            no_applicable_targets_behavior=NoApplicableTargetsBehavior.error,
            expect_single_field_set=True,
        ),
    )
    target = targets_to_valid_field_sets.targets[0]
    field_set = targets_to_valid_field_sets.field_sets[0]
    request = await Get(RunRequest, RunFieldSet, field_set)

    with temporary_dir(
        root_dir=global_options.options.pants_workdir, cleanup=True
    ) as tmpdir:
        path_prefix = PurePath(tmpdir).relative_to(build_root.path).as_posix()
        workspace.write_digest(
            request.digest,
            path_prefix=path_prefix,
        )

        args = (arg.format(chroot=tmpdir) for arg in request.args)
        env = {
            **complete_env,
            **{k: v.format(chroot=tmpdir) for k, v in request.extra_env.items()},
        }
        try:
            result = interactive_runner.run(
                InteractiveProcess(
                    argv=(*args, *django_run_subsystem.args),
                    env=env,
                    run_in_workspace=True,
                )
            )
            exit_code = result.exit_code
        except Exception as e:
            console.print_stderr(
                f"Exception when attempting to run {field_set.address}: {e!r}"
            )
            exit_code = -1
        else:
            mutable_files_field = target.get(MutableFiles)
            paths = [join(tmpdir, mutable_file_glob) for mutable_file_glob in mutable_files_field.value]
            mutated_file_globs = [glob.glob(p, recursive=True) for p in paths]
            contents = []
            for file_glob in mutated_file_globs:
                for path in file_glob:
                    write_path = relpath(path, tmpdir)
                    if exists(write_path):
                        continue
                    copyfile(path, write_path)

    return DjangoRun(exit_code=exit_code)


def target_types():
    return [DjangoAppTarget]


def rules():
    return collect_rules()

target_types.py

from pants.backend.python.target_types import (
    PexBinary,
    PexEntryPointField,
)
from pants.engine.target import StringSequenceField, Target
from pants.util.ordered_set import FrozenOrderedSet


class DjangoManagePyEntryPoint(PexEntryPointField):
    default = "manage.py"
    required = False


class MutableFiles(StringSequenceField):
    alias = "mutable_files"
    help = "The set of files that are mutable during the run of a dango command."


class DjangoAppTarget(Target):
    alias = "django_app"
    core_fields = (
        *(FrozenOrderedSet(PexBinary.core_fields) - {PexEntryPointField}),
        DjangoManagePyEntryPoint,
        MutableFiles,
    )
    help = "A django app target"