pantsbuild / pants / 19015773527

from dataclasses import dataclass

from pants.backend.python.lint.autoflake.skip_field import SkipAutoflakeField

from pants.backend.python.lint.autoflake.subsystem import Autoflake

from pants.backend.python.target_types import PythonSourceField

from pants.backend.python.util_rules import pex

from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex

from pants.core.goals.fix import FixResult, FixTargetsRequest

from pants.core.util_rules.partitions import PartitionerType

from pants.engine.process import execute_process_or_raise

from pants.engine.rules import collect_rules, implicitly, rule

from pants.engine.target import FieldSet, Target

from pants.util.logging import LogLevel

from pants.util.strutil import pluralize

@dataclass(frozen=True)

class AutoflakeFieldSet(FieldSet):

    required_fields = (PythonSourceField,)

    @classmethod

    def opt_out(cls, tgt: Target) -> bool:

        return tgt.get(SkipAutoflakeField).value

class AutoflakeRequest(FixTargetsRequest):

    field_set_type = AutoflakeFieldSet

    tool_subsystem = Autoflake  # type: ignore[assignment]

    partitioner_type = PartitionerType.DEFAULT_SINGLE_PARTITION

@rule(desc="Fix with Autoflake", level=LogLevel.DEBUG)

async def autoflake_fix(request: AutoflakeRequest.Batch, autoflake: Autoflake) -> FixResult:

    autoflake_pex = await create_venv_pex(**implicitly(autoflake.to_pex_request()))

    result = await execute_process_or_raise(

    return await FixResult.create(request, result)

def rules():

    return (