diff --git a/.github/scripts/summary.py b/.github/scripts/summary.py index d243065..819a4f6 100644 --- a/.github/scripts/summary.py +++ b/.github/scripts/summary.py @@ -1,8 +1,10 @@ # ruff: noqa:D100,D101,D102,D103,D105 +from __future__ import annotations + import argparse import logging from dataclasses import InitVar, dataclass, field -from typing import Any, List, Literal, Optional +from typing import Any, Literal from xml.etree import ElementTree from tabulate import tabulate @@ -16,10 +18,10 @@ class TestCase: name: str time: float state: Literal["success", "failure", "error", "skipped"] - summary: Optional[str] + summary: str | None @classmethod - def from_junit(cls, tree: ElementTree.Element) -> "TestCase": + def from_junit(cls, tree: ElementTree.Element) -> TestCase: children = tree.getchildren() assert len(children) <= 1 @@ -67,13 +69,13 @@ class TestSuite: failures: int skipped: int time: float - tests: List[TestCase] + tests: list[TestCase] def __post_init__(self, total: int) -> None: self.successes = total - self.errors - self.failures - self.skipped @classmethod - def from_junit(cls, tree: ElementTree.Element) -> "TestSuite": + def from_junit(cls, tree: ElementTree.Element) -> TestSuite: attrs = tree.attrib tests = [ @@ -97,7 +99,7 @@ def __lt__(self, other: Any) -> bool: return (self.time, self.name) < (other.time, other.name) -def get_failures_and_errors(testsuites: List[TestSuite]) -> str: +def get_failures_and_errors(testsuites: list[TestSuite]) -> str: reports = [] for testsuite in testsuites: diff --git a/docs/_extensions/cleanup_signatures.py b/docs/_extensions/cleanup_signatures.py index 153faa4..75a23a3 100644 --- a/docs/_extensions/cleanup_signatures.py +++ b/docs/_extensions/cleanup_signatures.py @@ -1,7 +1,10 @@ -from typing import Any, Dict, Optional, Tuple +from __future__ import annotations -from sphinx.application import Sphinx -from sphinx.ext.autodoc import Options +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from sphinx.application import Sphinx + from sphinx.ext.autodoc import Options # ruff: noqa: ARG001 @@ -11,9 +14,9 @@ def cleanup_signatures( # pylint: disable=unused-argument name: str, obj: Any, options: Options, - signature: Optional[str], - return_annotation: Optional[str], -) -> Optional[Tuple[str, Optional[str]]]: + signature: str | None, + return_annotation: str | None, +) -> tuple[str, str | None] | None: if name == "dwas.StepRunner": # Hide the __init__ signature for dwas.StepRunner, it's meant to be # private @@ -21,7 +24,7 @@ def cleanup_signatures( # pylint: disable=unused-argument return None -def setup(app: Sphinx) -> Dict[str, Any]: +def setup(app: Sphinx) -> dict[str, Any]: app.connect("autodoc-process-signature", cleanup_signatures) return { diff --git a/docs/_extensions/execute.py b/docs/_extensions/execute.py index 21d4040..ae7bbf5 100644 --- a/docs/_extensions/execute.py +++ b/docs/_extensions/execute.py @@ -1,15 +1,19 @@ +from __future__ import annotations + import os import shlex import subprocess -from typing import Any, Dict, List +from typing import TYPE_CHECKING, Any from docutils import nodes from docutils.parsers import rst from docutils.parsers.rst.directives import nonnegative_int, unchanged -from sphinx.addnodes import document -from sphinx.application import Sphinx from sphinx.errors import ExtensionError +if TYPE_CHECKING: + from sphinx.addnodes import document + from sphinx.application import Sphinx + class execute(nodes.Element): # pylint: disable=invalid-name # noqa: N801 pass @@ -25,7 +29,7 @@ class ExecuteDirective(rst.Directive): "cwd": unchanged, } - def run(self) -> List[nodes.Element]: + def run(self) -> list[nodes.Element]: env = self.state.document.settings.env node = execute() @@ -77,7 +81,7 @@ def run_programs( node.replace_self(new_node) -def setup(app: Sphinx) -> Dict[str, Any]: +def setup(app: Sphinx) -> dict[str, Any]: app.add_directive("command-output", ExecuteDirective) app.connect("doctree-read", run_programs) return {"parallel_read_safe": True} diff --git a/pyproject.toml b/pyproject.toml index b55d000..ecab143 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -178,11 +178,6 @@ ignore = [ "S603", # We want to be able tor rely on PATH and not hardcode binaries "S607", - - # FIXME: re-enable future annotations - "FA100", - # FIXME: re-enable upgrade rules - "UP", ] flake8-pytest-style.parametrize-values-type = "tuple" diff --git a/src/dwas/__main__.py b/src/dwas/__main__.py index b75b133..b2bd61b 100644 --- a/src/dwas/__main__.py +++ b/src/dwas/__main__.py @@ -1,4 +1,6 @@ # ruff: noqa:D100,D101,D102,D103 +from __future__ import annotations + import importlib.util import logging import os @@ -14,13 +16,15 @@ ) from contextvars import copy_context from importlib.metadata import version -from typing import Any, Dict, List, Optional +from typing import TYPE_CHECKING, Any from . import _io, _pipeline from ._config import Config from ._exceptions import BaseDwasException, FailedPipelineException from ._logging import setup_logging -from ._steps.handlers import BaseStepHandler + +if TYPE_CHECKING: + from ._steps.handlers import BaseStepHandler LOGGER = logging.getLogger(__name__) @@ -31,7 +35,7 @@ def __call__( parser: ArgumentParser, # noqa:ARG002 namespace: Namespace, values: Any, - option_string: Optional[str] = None, # noqa:ARG002 + option_string: str | None = None, # noqa:ARG002 ) -> None: items = getattr(namespace, self.dest, None) if items is None: @@ -52,7 +56,7 @@ def __call__( parser: ArgumentParser, # noqa:ARG002 namespace: Namespace, values: Any, # noqa:ARG002 - option_string: Optional[str] = None, + option_string: str | None = None, ) -> None: assert option_string is not None @@ -65,7 +69,7 @@ def format_usage(self) -> str: return " | ".join(self.option_strings) -def _parse_args(args: Optional[List[str]] = None) -> Namespace: +def _parse_args(args: list[str] | None = None) -> Namespace: parser = ArgumentParser( formatter_class=RawDescriptionHelpFormatter, epilog="""\ @@ -195,8 +199,8 @@ def _parse_args(args: Optional[List[str]] = None) -> Namespace: def _parse_steps( - args: List[str], known_steps: Dict[str, BaseStepHandler] -) -> Optional[List[str]]: + args: list[str], known_steps: dict[str, BaseStepHandler] +) -> list[str] | None: if not args: return None @@ -275,8 +279,8 @@ def _load_user_config( def _execute_pipeline( config: Config, pipeline_config: str, - steps_parameters: List[str], - except_steps: Optional[List[str]], + steps_parameters: list[str], + except_steps: list[str] | None, *, only_selected_step: bool, clean: bool, @@ -314,7 +318,7 @@ def _execute_pipeline( @_io.instrument_streams() -def main(sys_args: Optional[List[str]] = None) -> None: +def main(sys_args: list[str] | None = None) -> None: if sys_args is None: sys_args = sys.argv[1:] if env_args := os.environ.get("DWAS_ADDOPTS"): diff --git a/src/dwas/_config.py b/src/dwas/_config.py index cc34c1e..5be1e4e 100644 --- a/src/dwas/_config.py +++ b/src/dwas/_config.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import logging import multiprocessing import os @@ -5,7 +7,6 @@ import shutil import sys from pathlib import Path -from typing import Dict, Optional from ._exceptions import BaseDwasException @@ -56,7 +57,7 @@ class Config: - Finally, it will look if this is attached to a tty and enable colors if so. """ - environ: Dict[str, str] + environ: dict[str, str] """ The environment to use when running commands. @@ -123,9 +124,9 @@ class Config: def __init__( self, cache_path: str, - log_path: Optional[str], + log_path: str | None, verbosity: int, - colors: Optional[bool], + colors: bool | None, n_jobs: int, *, skip_missing_interpreters: bool, @@ -201,7 +202,7 @@ def __init__( self.environ["PY_COLORS"] = "0" self.environ["NO_COLOR"] = "0" - def _get_color_setting(self, colors: Optional[bool]) -> bool: + def _get_color_setting(self, colors: bool | None) -> bool: # pylint: disable=too-many-return-statements if colors is not None: return colors diff --git a/src/dwas/_dependency_injection.py b/src/dwas/_dependency_injection.py index f8a8b47..1f0dbf4 100644 --- a/src/dwas/_dependency_injection.py +++ b/src/dwas/_dependency_injection.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import inspect import logging -from typing import Any, Callable, Dict, TypeVar +from typing import Any, Callable, TypeVar from ._exceptions import BaseDwasException from ._inspect import get_location, get_name @@ -10,7 +12,7 @@ def call_with_parameters( - func: Callable[..., T], parameters: Dict[str, Any] + func: Callable[..., T], parameters: dict[str, Any] ) -> T: signature = inspect.signature(func) kwargs = {} diff --git a/src/dwas/_exceptions.py b/src/dwas/_exceptions.py index f305386..ae33d81 100644 --- a/src/dwas/_exceptions.py +++ b/src/dwas/_exceptions.py @@ -1,4 +1,4 @@ -from typing import List +from __future__ import annotations class BaseDwasException(Exception): @@ -46,13 +46,13 @@ def _pluralize(self, n_jobs: int) -> str: class UnknownStepsException(BaseDwasException): - def __init__(self, steps: List[str]) -> None: + def __init__(self, steps: list[str]) -> None: message = f"Unknown steps: {', '.join(steps)}" super().__init__(message) class CyclicStepDependenciesException(BaseDwasException): - def __init__(self, cycle: List[str]) -> None: + def __init__(self, cycle: list[str]) -> None: message = f"Cyclic dependencies between steps: {' --> '.join(cycle)}" super().__init__(message) self.cycle = cycle diff --git a/src/dwas/_frontend.py b/src/dwas/_frontend.py index 121f358..04fc724 100644 --- a/src/dwas/_frontend.py +++ b/src/dwas/_frontend.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import shutil import sys import time @@ -5,14 +7,16 @@ from contextvars import copy_context from datetime import timedelta from threading import Event, Thread -from typing import Iterator, List +from typing import TYPE_CHECKING, Iterator from colorama import Cursor, Fore, Style, ansi from . import _io -from ._scheduler import Scheduler from ._timing import format_timedelta +if TYPE_CHECKING: + from ._scheduler import Scheduler + ANSI_SHOW_CURSOR = f"{ansi.CSI}?25h" ANSI_HIDE_CURSOR = f"{ansi.CSI}?25l" @@ -25,7 +29,7 @@ def __init__(self, scheduler: Scheduler, start_time: float) -> None: def _counter(self, value: int, color: str) -> str: return f"{color}{Style.BRIGHT}{value}{Style.NORMAL}{Fore.YELLOW}" - def lines(self) -> List[str]: + def lines(self) -> list[str]: update_at = time.monotonic() term_width = shutil.get_terminal_size().columns @@ -55,7 +59,7 @@ def lines(self) -> List[str]: "~", ) - additional_info: List[str] = [] + additional_info: list[str] = [] if self._scheduler.ready: ready_line = ( f"[-:--:--] {Fore.YELLOW}{Style.BRIGHT}ready: " diff --git a/src/dwas/_io.py b/src/dwas/_io.py index 9a06624..b422df7 100644 --- a/src/dwas/_io.py +++ b/src/dwas/_io.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import io import re import sys @@ -10,8 +12,10 @@ suppress, ) from contextvars import ContextVar -from pathlib import Path -from typing import Deque, Generator, Iterator, Optional, TextIO, Tuple +from typing import TYPE_CHECKING, Generator, Iterator, TextIO + +if TYPE_CHECKING: + from pathlib import Path STDOUT = ContextVar[TextIO]("STDOUT") STDERR = ContextVar[TextIO]("STDERR") @@ -24,7 +28,7 @@ class NoOpWriter(io.TextIOWrapper): def __init__(self) -> None: pass - def read(self, size: Optional[int] = None) -> str: # noqa: ARG002 + def read(self, size: int | None = None) -> str: # noqa: ARG002 raise io.UnsupportedOperation("Can't read from a noopwriter") def write(self, data: str) -> int: @@ -37,11 +41,11 @@ def flush(self) -> None: class MemoryPipe(io.TextIOWrapper): def __init__( self, - writer: "PipePlexer", + writer: PipePlexer, ) -> None: self._writer = writer - def read(self, size: Optional[int] = None) -> str: # noqa: ARG002 + def read(self, size: int | None = None) -> str: # noqa: ARG002 raise io.UnsupportedOperation("can't read from a memorypipe") def write(self, data: str) -> int: @@ -56,7 +60,7 @@ def __init__(self, *, write_on_flush: bool = True) -> None: self.stderr = MemoryPipe(self) self.stdout = MemoryPipe(self) - self._buffer: Deque[Tuple[MemoryPipe, str]] = deque() + self._buffer: deque[tuple[MemoryPipe, str]] = deque() self._write_on_flush = write_on_flush def write(self, stream: MemoryPipe, data: str) -> int: @@ -65,7 +69,7 @@ def write(self, stream: MemoryPipe, data: str) -> int: def flush( self, force_write: bool = False # noqa:FBT001,FBT002 - ) -> Optional[int]: + ) -> int | None: line = None if self._write_on_flush or force_write: @@ -96,7 +100,7 @@ def __init__( self._var = var self._log_var = log_var - def read(self, size: Optional[int] = None) -> str: # noqa: ARG002 + def read(self, size: int | None = None) -> str: # noqa: ARG002 raise io.UnsupportedOperation("can't read from a memorypipe") def write(self, data: str) -> int: @@ -139,7 +143,7 @@ def redirect_streams(stdout: TextIO, stderr: TextIO) -> Iterator[None]: @contextmanager -def log_file(path: Optional[Path]) -> Iterator[None]: +def log_file(path: Path | None) -> Iterator[None]: with ExitStack() as stack: if path is None: fd: TextIO = NoOpWriter() diff --git a/src/dwas/_logging.py b/src/dwas/_logging.py index 9ac41a3..9e56993 100644 --- a/src/dwas/_logging.py +++ b/src/dwas/_logging.py @@ -1,12 +1,16 @@ +from __future__ import annotations + import logging -from contextvars import ContextVar from types import MappingProxyType, TracebackType -from typing import Any, Optional, TextIO, Tuple, Type, Union, cast +from typing import TYPE_CHECKING, Any, TextIO, cast from colorama import Back, Fore, Style, init from ._io import ANSI_ESCAPE_CODE_RE +if TYPE_CHECKING: + from contextvars import ContextVar + class ColorFormatter(logging.Formatter): # We need to follow camel case style @@ -28,10 +32,8 @@ def formatMessage(self, record: logging.LogRecord) -> str: def formatException( self, - ei: Union[ - Tuple[Type[BaseException], BaseException, Optional[TracebackType]], - Tuple[None, None, None], - ], + ei: tuple[type[BaseException], BaseException, TracebackType | None] + | tuple[None, None, None], ) -> str: output = super().formatException(ei) return f"{Fore.CYAN}\ndwas > " + "\ndwas > ".join(output.splitlines()) diff --git a/src/dwas/_pipeline.py b/src/dwas/_pipeline.py index 8a5b4cb..ad46c93 100644 --- a/src/dwas/_pipeline.py +++ b/src/dwas/_pipeline.py @@ -9,17 +9,7 @@ from contextvars import ContextVar, copy_context from datetime import timedelta from subprocess import CalledProcessError -from typing import ( - TYPE_CHECKING, - Callable, - Dict, - Generator, - List, - Optional, - Set, - Tuple, - cast, -) +from typing import TYPE_CHECKING, Callable, Generator, cast from colorama import Fore, Style @@ -49,11 +39,11 @@ _PIPELINE = ContextVar["Pipeline"]("pipeline") -def get_pipeline() -> "Pipeline": +def get_pipeline() -> Pipeline: return _PIPELINE.get() -def set_pipeline(pipeline: "Pipeline") -> None: +def set_pipeline(pipeline: Pipeline) -> None: _PIPELINE.set(pipeline) @@ -71,35 +61,35 @@ def __init__(self, config: Config) -> None: self.config = config self.proc_manager = ProcessManager() - self._registered_steps: List[Tuple[str, Step, Optional[str]]] = [] - self._registered_step_groups: List[ - Tuple[str, List[str], Optional[str], Optional[bool]] + self._registered_steps: list[tuple[str, Step, str | None]] = [] + self._registered_step_groups: list[ + tuple[str, list[str], str | None, bool | None] ] = [] - self._steps_cache: Optional[Dict[str, BaseStepHandler]] = None + self._steps_cache: dict[str, BaseStepHandler] | None = None @property - def steps(self) -> Dict[str, BaseStepHandler]: + def steps(self) -> dict[str, BaseStepHandler]: if self._steps_cache is None: self._steps_cache = self._resolve_steps() return self._steps_cache def register_step( - self, name: str, description: Optional[str], step: "Step" + self, name: str, description: str | None, step: Step ) -> None: self._registered_steps.append((name, step, description)) def register_step_group( self, name: str, - requires: List[str], - description: Optional[str] = None, - run_by_default: Optional[bool] = None, + requires: list[str], + description: str | None = None, + run_by_default: bool | None = None, ) -> None: self._registered_step_groups.append( (name, requires, description, run_by_default) ) - def _resolve_steps(self) -> Dict[str, BaseStepHandler]: + def _resolve_steps(self) -> dict[str, BaseStepHandler]: steps = {} for name, func, description in self._registered_steps: @@ -125,7 +115,7 @@ def _resolve_steps(self) -> Dict[str, BaseStepHandler]: return steps def _resolve_parameters( - self, name: str, func: Step, description: Optional[str] + self, name: str, func: Step, description: str | None ) -> Generator[BaseStepHandler, None, None]: parameters = extract_parameters(func) all_run_by_default = True @@ -170,17 +160,17 @@ def _resolve_parameters( def _build_graph( # noqa: C901 self, - steps: Optional[List[str]] = None, - except_steps: Optional[List[str]] = None, + steps: list[str] | None = None, + except_steps: list[str] | None = None, *, only_selected_steps: bool = False, - ) -> Dict[str, List[str]]: + ) -> dict[str, list[str]]: # we should refactor at some point # pylint: disable=too-many-branches,too-many-locals,too-many-statements # First build the whole graph, without ignoring edges. This is necessary # to ensure we keep all dependency relations - def expand(steps: List[str]) -> Set[str]: + def expand(steps: list[str]) -> set[str]: expanded = set() to_process = deque(steps) while to_process: @@ -228,10 +218,10 @@ def expand(steps: List[str]) -> Set[str]: raise UnknownStepsException(unknown_steps) if except_steps_set: - except_replacements: Dict[str, List[str]] = {} + except_replacements: dict[str, list[str]] = {} # FIXME: ensure we handle cycles - def compute_replacement(requirements: List[str]) -> List[str]: + def compute_replacement(requirements: list[str]) -> list[str]: replacements = [] for requirement in requirements: if requirement in steps: @@ -266,8 +256,8 @@ def compute_replacement(requirements: List[str]) -> List[str]: only_replacements = {} def compute_only_replacement( - step: str, requirements: List[str] - ) -> List[str]: + step: str, requirements: list[str] + ) -> list[str]: if step in only: return [step] @@ -299,8 +289,8 @@ def compute_only_replacement( def execute( self, - steps: Optional[List[str]], - except_steps: Optional[List[str]], + steps: list[str] | None, + except_steps: list[str] | None, *, only_selected_steps: bool, clean: bool, @@ -324,7 +314,7 @@ def execute( should_stop = False - def request_stop(_signum: int, _frame: Optional[FrameType]) -> None: + def request_stop(_signum: int, _frame: FrameType | None) -> None: nonlocal should_stop if not should_stop: @@ -358,8 +348,8 @@ def get_step(self, step_name: str) -> BaseStepHandler: def list_all_steps( self, - steps: Optional[List[str]] = None, - except_steps: Optional[List[str]] = None, + steps: list[str] | None = None, + except_steps: list[str] | None = None, *, only_selected_steps: bool = False, show_dependencies: bool = False, @@ -425,8 +415,8 @@ def _execute( scheduler: Scheduler, should_stop: Callable[[], bool], ) -> None: - running_futures: Dict[ - futures.Future[None], Tuple[str, Optional[_io.PipePlexer]] + running_futures: dict[ + futures.Future[None], tuple[str, _io.PipePlexer | None] ] = {} with futures.ThreadPoolExecutor(self.config.n_jobs) as executor: @@ -527,8 +517,8 @@ def _handle_step_result( def _log_summary( self, scheduler: Scheduler, - steps_order: List[str], - graph: Dict[str, List[str]], + steps_order: list[str], + graph: dict[str, list[str]], start_time: float, ) -> None: LOGGER.info("%s*** Steps summary ***", Style.BRIGHT) @@ -607,7 +597,7 @@ def _log_summary( def _run_step( self, name: str, - pipe_plexer: Optional[_io.PipePlexer], + pipe_plexer: _io.PipePlexer | None, ) -> None: with ExitStack() as stack: if pipe_plexer is not None: @@ -634,8 +624,8 @@ def _format_exception(self, exc: Exception) -> str: def _display_slowest_dependency_chain( self, - graph: Dict[str, List[str]], - results: Dict[str, Tuple[JobResult, timedelta, Optional[Exception]]], + graph: dict[str, list[str]], + results: dict[str, tuple[JobResult, timedelta, Exception | None]], ) -> None: if len(graph) <= 1: # If there's a single entry in the whole graph, no need to show @@ -669,12 +659,12 @@ def _display_slowest_dependency_chain( def _compute_slowest_chains( self, - graph: Dict[str, List[str]], - results: Dict[str, Tuple[JobResult, timedelta, Optional[Exception]]], - ) -> Dict[str, Tuple[List[str], timedelta]]: - total_time_per_step: Dict[str, Tuple[List[str], timedelta]] = {} + graph: dict[str, list[str]], + results: dict[str, tuple[JobResult, timedelta, Exception | None]], + ) -> dict[str, tuple[list[str], timedelta]]: + total_time_per_step: dict[str, tuple[list[str], timedelta]] = {} - def compute_chain(step: str) -> Tuple[List[str], timedelta]: + def compute_chain(step: str) -> tuple[list[str], timedelta]: precomputed_result = total_time_per_step.get(step, None) if precomputed_result is not None: return precomputed_result diff --git a/src/dwas/_runners.py b/src/dwas/_runners.py index 97f8430..a2d3b4e 100644 --- a/src/dwas/_runners.py +++ b/src/dwas/_runners.py @@ -5,7 +5,7 @@ import shutil import sys from contextlib import suppress -from typing import TYPE_CHECKING, Dict, List, Optional, cast +from typing import TYPE_CHECKING, cast from virtualenv import session_via_cli @@ -30,7 +30,7 @@ class _VirtualenvInstaller: - def __init__(self, path: Path, python_spec: Optional[str]) -> None: + def __init__(self, path: Path, python_spec: str | None) -> None: self.environ = os.environ.copy() self.environ["VIRTUALENV_CLEAR"] = "False" if python_spec is not None: @@ -46,7 +46,7 @@ def python(self) -> str: return str(self._session.creator.exe) @property - def paths(self) -> List[str]: + def paths(self) -> list[str]: creator = self._session.creator if creator.bin_dir == creator.script_dir: return [str(creator.bin_dir)] @@ -81,9 +81,9 @@ class VenvRunner: def __init__( self, name: str, - python_spec: Optional[str], + python_spec: str | None, config: Config, - environ: Dict[str, str], + environ: dict[str, str], proc_manager: ProcessManager, ) -> None: self._path = (config.venvs_path / name.replace(":", "-")).resolve() @@ -122,11 +122,9 @@ def install(self, *packages: str) -> None: def run( self, - command: List[str], - cwd: Optional[ - str | bytes | os.PathLike[str] | os.PathLike[bytes] - ] = None, - env: Optional[Dict[str, str]] = None, + command: list[str], + cwd: str | bytes | os.PathLike[str] | os.PathLike[bytes] | None = None, + env: dict[str, str] | None = None, *, external_command: bool = False, silent_on_success: bool = False, @@ -144,8 +142,8 @@ def run( ) def _merge_env( - self, config: Config, additional_env: Optional[Dict[str, str]] = None - ) -> Dict[str, str]: + self, config: Config, additional_env: dict[str, str] | None = None + ) -> dict[str, str]: env = config.environ.copy() env.update(self._environ) env.update( @@ -159,7 +157,7 @@ def _merge_env( return env def _validate_command( - self, command: str, env: Dict[str, str], *, external_command: bool + self, command: str, env: dict[str, str], *, external_command: bool ) -> None: cmd = shutil.which(command, path=env["PATH"]) diff --git a/src/dwas/_scheduler.py b/src/dwas/_scheduler.py index 6572c48..c0ee2c3 100644 --- a/src/dwas/_scheduler.py +++ b/src/dwas/_scheduler.py @@ -1,20 +1,12 @@ +from __future__ import annotations + import copy import enum import logging import time from collections import deque from datetime import timedelta -from typing import ( - Any, - Deque, - Dict, - Iterable, - List, - Mapping, - Optional, - Set, - Tuple, -) +from typing import Any, Iterable, Mapping from ._exceptions import CyclicStepDependenciesException @@ -34,26 +26,26 @@ class Scheduler: # pylint: disable=too-many-instance-attributes def __init__( self, - dependencies_graph: Dict[str, Set[str]], - dependents_graph: Dict[str, Set[str]], - weights: Dict[str, Any], + dependencies_graph: dict[str, set[str]], + dependents_graph: dict[str, set[str]], + weights: dict[str, Any], ) -> None: self._dependencies_graph = copy.deepcopy(dependencies_graph) self._dependents_graph = copy.deepcopy(dependents_graph) self._weights = weights self._stopped = False - self.waiting: Set[str] = set() - self.ready: List[str] = [] - self.running: Dict[str, float] = {} - self.success: Set[str] = set() - self.failed: Set[str] = set() - self.blocked: Set[str] = set() - self.cancelled: Set[str] = set() - self.skipped: Set[str] = set() - - self.results: Dict[ - str, Tuple[JobResult, timedelta, Optional[Exception]] + self.waiting: set[str] = set() + self.ready: list[str] = [] + self.running: dict[str, float] = {} + self.success: set[str] = set() + self.failed: set[str] = set() + self.blocked: set[str] = set() + self.cancelled: set[str] = set() + self.skipped: set[str] = set() + + self.results: dict[ + str, tuple[JobResult, timedelta, Exception | None] ] = {} # Enqueue all steps that are ready @@ -165,8 +157,8 @@ def __init__(self, graph: Mapping[str, Iterable[str]]) -> None: # as a way to sort. self._weights = self._build_weights() - def _make_dependent_graph(self) -> Dict[str, Set[str]]: - graph: Dict[str, Set[str]] = { + def _make_dependent_graph(self) -> dict[str, set[str]]: + graph: dict[str, set[str]] = { key: set() for key in self._dependencies_graph } @@ -176,13 +168,13 @@ def _make_dependent_graph(self) -> Dict[str, Set[str]]: return graph - def _build_weights(self) -> Dict[str, Tuple[int, int, str]]: - weights: Dict[str, Tuple[int, int, str]] = {} + def _build_weights(self) -> dict[str, tuple[int, int, str]]: + weights: dict[str, tuple[int, int, str]] = {} - path: Deque[str] = deque() + path: deque[str] = deque() path_set = set() - def _compute_weight(step: str) -> Tuple[int, int, str]: + def _compute_weight(step: str) -> tuple[int, int, str]: if step in weights: return weights[step] @@ -214,7 +206,7 @@ def get_scheduler(self) -> Scheduler: self._dependencies_graph, self._dependents_graph, self._weights ) - def order(self) -> List[str]: + def order(self) -> list[str]: entries = [] scheduler = self.get_scheduler() diff --git a/src/dwas/_steps/handlers.py b/src/dwas/_steps/handlers.py index 7fd0e05..05f6ba3 100644 --- a/src/dwas/_steps/handlers.py +++ b/src/dwas/_steps/handlers.py @@ -7,7 +7,7 @@ import shutil from abc import ABC, abstractmethod from contextlib import suppress -from typing import TYPE_CHECKING, Any, Dict, List, Optional +from typing import TYPE_CHECKING, Any from .._dependency_injection import call_with_parameters from .._exceptions import BaseDwasException @@ -35,10 +35,10 @@ class BaseStepHandler(ABC): def __init__( self, name: str, - pipeline: "Pipeline", - requires: Optional[List[str]] = None, - run_by_default: Optional[bool] = None, - description: Optional[str] = None, + pipeline: Pipeline, + requires: list[str] | None = None, + run_by_default: bool | None = None, + description: str | None = None, ) -> None: self.name = name self.description = description @@ -57,17 +57,15 @@ def execute(self) -> None: pass @abstractmethod - def set_user_args(self, args: List[str]) -> None: + def set_user_args(self, args: list[str]) -> None: pass @abstractmethod - def _execute_dependent_setup( - self, current_step: "BaseStepHandler" - ) -> None: + def _execute_dependent_setup(self, current_step: BaseStepHandler) -> None: pass @abstractmethod - def _get_artifacts(self, key: str) -> List[Any]: + def _get_artifacts(self, key: str) -> list[Any]: pass @@ -76,14 +74,14 @@ def __init__( self, name: str, func: Step, - pipeline: "Pipeline", - python_spec: Optional[str], - requires: Optional[List[str]] = None, - run_by_default: Optional[bool] = None, - description: Optional[str] = None, - parameters: Optional[Dict[str, Any]] = None, - passenv: Optional[List[str]] = None, - setenv: Optional[Dict[str, str]] = None, + pipeline: Pipeline, + python_spec: str | None, + requires: list[str] | None = None, + run_by_default: bool | None = None, + description: str | None = None, + parameters: dict[str, Any] | None = None, + passenv: list[str] | None = None, + setenv: dict[str, str] | None = None, ) -> None: super().__init__(name, pipeline, requires, run_by_default, description) @@ -101,7 +99,7 @@ def __init__( self._step_runner = StepRunner(self) if parameters is None: - self.parameters: Dict[str, Any] = { + self.parameters: dict[str, Any] = { "step": self._step_runner, "user_args": None, } @@ -126,7 +124,7 @@ def config(self) -> Config: def python(self) -> str: return self._venv_runner.python - def set_user_args(self, args: List[str]) -> None: + def set_user_args(self, args: list[str]) -> None: step_signature = inspect.signature(self._func) if "user_args" not in step_signature.parameters: LOGGER.warning( @@ -135,7 +133,7 @@ def set_user_args(self, args: List[str]) -> None: ) self.parameters["user_args"] = args - def get_artifacts(self, key: str) -> List[Any]: + def get_artifacts(self, key: str) -> list[Any]: return list( itertools.chain.from_iterable( [ @@ -154,12 +152,10 @@ def install(self, *packages: str) -> None: def run( self, - command: List[str], + command: list[str], *, - cwd: Optional[ - str | bytes | os.PathLike[str] | os.PathLike[bytes] - ] = None, - env: Optional[Dict[str, str]] = None, + cwd: str | bytes | os.PathLike[str] | os.PathLike[bytes] | None = None, + env: dict[str, str] | None = None, external_command: bool = False, silent_on_success: bool = False, ) -> subprocess.CompletedProcess[None]: @@ -202,9 +198,7 @@ def clean(self) -> None: with suppress(FileNotFoundError): shutil.rmtree(self._step_runner.cache_path) - def _execute_dependent_setup( - self, current_step: "BaseStepHandler" - ) -> None: + def _execute_dependent_setup(self, current_step: BaseStepHandler) -> None: assert isinstance(current_step, StepHandler) if isinstance(self._func, StepWithDependentSetup): @@ -216,7 +210,7 @@ def _execute_dependent_setup( current_step._step_runner, # noqa: SLF001 ) - def _get_artifacts(self, key: str) -> List[Any]: + def _get_artifacts(self, key: str) -> list[Any]: if isinstance(self._func, StepWithArtifacts): all_artifacts = self._func.gather_artifacts(self._step_runner) else: @@ -233,8 +227,8 @@ def _get_artifacts(self, key: str) -> List[Any]: return artifacts def _resolve_environ( - self, passenv: Optional[List[str]], setenv: Optional[Dict[str, str]] - ) -> Dict[str, str]: + self, passenv: list[str] | None, setenv: dict[str, str] | None + ) -> dict[str, str]: base_env = {} if setenv: base_env.update(setenv) @@ -271,13 +265,11 @@ def clean(self) -> None: def execute(self) -> None: LOGGER.debug("Step %s is a meta step. Nothing to do", self.name) - def set_user_args(self, args: List[str]) -> None: + def set_user_args(self, args: list[str]) -> None: for requirement in self.requires: self._pipeline.get_step(requirement).set_user_args(args) - def _execute_dependent_setup( - self, current_step: "BaseStepHandler" - ) -> None: + def _execute_dependent_setup(self, current_step: BaseStepHandler) -> None: for requirement in self.requires: # Pylint check here is wrong, it's still an instance of our class # pylint: disable=protected-access @@ -285,7 +277,7 @@ def _execute_dependent_setup( requirement )._execute_dependent_setup(current_step) - def _get_artifacts(self, key: str) -> List[Any]: + def _get_artifacts(self, key: str) -> list[Any]: return list( itertools.chain.from_iterable( [ diff --git a/src/dwas/_steps/parametrize.py b/src/dwas/_steps/parametrize.py index 74aaee7..e89ef0d 100644 --- a/src/dwas/_steps/parametrize.py +++ b/src/dwas/_steps/parametrize.py @@ -1,16 +1,7 @@ +from __future__ import annotations + import itertools -from typing import ( - Any, - Callable, - Dict, - List, - Optional, - Sequence, - Tuple, - TypeVar, - Union, - overload, -) +from typing import Any, Callable, Sequence, TypeVar, overload from .._exceptions import BaseDwasException @@ -56,18 +47,18 @@ def __init__(self, n_args_values: int, n_args_ids: int) -> None: class Parameter: - def __init__(self, id_: Optional[str], parameters: Dict[str, Any]) -> None: + def __init__(self, id_: str | None, parameters: dict[str, Any]) -> None: self._parameters = parameters if id_ is None: id_ = ",".join(str(v) for v in parameters.values()) self.id = id_ - def as_dict(self) -> Dict[str, Any]: + def as_dict(self) -> dict[str, Any]: return self._parameters.copy() # pylint: disable=protected-access @classmethod - def merge(cls, param1: "Parameter", param2: "Parameter") -> "Parameter": + def merge(cls, param1: Parameter, param2: Parameter) -> Parameter: # ruff: noqa: SLF001 if param1.id == "": id_ = param2.id @@ -98,7 +89,7 @@ def __repr__(self) -> str: def parametrize( arg_names: str, args_values: Sequence[Any], - ids: Optional[Sequence[Optional[str]]] = None, + ids: Sequence[str | None] | None = None, ) -> Callable[[T], T]: ... @@ -107,15 +98,15 @@ def parametrize( def parametrize( arg_names: Sequence[str], args_values: Sequence[Sequence[Any]], - ids: Optional[Sequence[Optional[str]]] = None, + ids: Sequence[str | None] | None = None, ) -> Callable[[T], T]: ... def parametrize( - arg_names: Union[str, Sequence[str]], - args_values: Union[Sequence[Any], Sequence[Sequence[Any]]], - ids: Optional[Sequence[Optional[str]]] = None, + arg_names: str | Sequence[str], + args_values: Sequence[Any] | Sequence[Sequence[Any]], + ids: Sequence[str | None] | None = None, ) -> Callable[[T], T]: """ Parametrize the decorated :term:`step` with the provided values. @@ -246,7 +237,7 @@ def _apply(func: T) -> T: return _apply -def set_defaults(values: Dict[str, Any]) -> Callable[[T], T]: +def set_defaults(values: dict[str, Any]) -> Callable[[T], T]: """ Set default values for parameters on the given :term:`step`. @@ -314,10 +305,10 @@ def build_parameters(**kwargs: Any) -> Callable[[T], T]: def extract_parameters( func: Callable[..., Any] -) -> List[Tuple[str, Dict[str, Any]]]: +) -> list[tuple[str, dict[str, Any]]]: defaults = getattr(func, _DEFAULTS, {}) - def _merge(parameter: Parameter) -> Dict[str, Any]: + def _merge(parameter: Parameter) -> dict[str, Any]: params = defaults.copy() params.update(parameter.as_dict()) return params diff --git a/src/dwas/_steps/registration.py b/src/dwas/_steps/registration.py index 8ae2184..19efb55 100644 --- a/src/dwas/_steps/registration.py +++ b/src/dwas/_steps/registration.py @@ -1,22 +1,26 @@ -from typing import Callable, Dict, List, Optional, Sequence +from __future__ import annotations + +from typing import TYPE_CHECKING, Callable, Sequence from .._exceptions import BaseDwasException from .._inspect import get_location from .._pipeline import get_pipeline from .parametrize import build_parameters, parametrize -from .steps import Step, StepRunner + +if TYPE_CHECKING: + from .steps import Step, StepRunner def register_step( func: Step, *, - name: Optional[str] = None, - description: Optional[str] = None, - python: Optional[str] = None, - requires: Optional[List[str]] = None, - run_by_default: Optional[bool] = None, - passenv: Optional[List[str]] = None, - setenv: Optional[Dict[str, str]] = None, + name: str | None = None, + description: str | None = None, + python: str | None = None, + requires: list[str] | None = None, + run_by_default: bool | None = None, + passenv: list[str] | None = None, + setenv: dict[str, str] | None = None, ) -> Step: """ Register the provided :term:`step`. @@ -155,15 +159,15 @@ def my_step(step: StepRunner): ... def register_managed_step( func: Step, - dependencies: Optional[Sequence[str]] = None, + dependencies: Sequence[str] | None = None, *, - name: Optional[str] = None, - description: Optional[str] = None, - python: Optional[str] = None, - requires: Optional[List[str]] = None, - run_by_default: Optional[bool] = None, - passenv: Optional[List[str]] = None, - setenv: Optional[Dict[str, str]] = None, + name: str | None = None, + description: str | None = None, + python: str | None = None, + requires: list[str] | None = None, + run_by_default: bool | None = None, + passenv: list[str] | None = None, + setenv: dict[str, str] | None = None, ) -> Step: """ Register the provided :term:`step`, and handle installing its dependencies. @@ -224,9 +228,9 @@ def install(step: StepRunner, dependencies: str) -> None: def register_step_group( name: str, - requires: List[str], - description: Optional[str] = None, - run_by_default: Optional[bool] = None, + requires: list[str], + description: str | None = None, + run_by_default: bool | None = None, ) -> None: """ Register a :term:`step group`. @@ -251,13 +255,13 @@ def register_step_group( def step( *, - name: Optional[str] = None, - description: Optional[str] = None, - python: Optional[str] = None, - requires: Optional[List[str]] = None, - run_by_default: Optional[bool] = None, - passenv: Optional[List[str]] = None, - setenv: Optional[Dict[str, str]] = None, + name: str | None = None, + description: str | None = None, + python: str | None = None, + requires: list[str] | None = None, + run_by_default: bool | None = None, + passenv: list[str] | None = None, + setenv: dict[str, str] | None = None, ) -> Callable[[Step], Step]: """ Register the decorated :term:`step` and make it available to the pipeline. @@ -294,13 +298,13 @@ def wrapper(func: Step) -> Step: def managed_step( dependencies: Sequence[str], *, - name: Optional[str] = None, - description: Optional[str] = None, - python: Optional[str] = None, - requires: Optional[List[str]] = None, - run_by_default: Optional[bool] = None, - passenv: Optional[List[str]] = None, - setenv: Optional[Dict[str, str]] = None, + name: str | None = None, + description: str | None = None, + python: str | None = None, + requires: list[str] | None = None, + run_by_default: bool | None = None, + passenv: list[str] | None = None, + setenv: dict[str, str] | None = None, ) -> Callable[[Step], Step]: """ Register the decorated :term:`step`, and handle installing its dependencies. diff --git a/src/dwas/_steps/steps.py b/src/dwas/_steps/steps.py index 1167018..7fb3df5 100644 --- a/src/dwas/_steps/steps.py +++ b/src/dwas/_steps/steps.py @@ -4,16 +4,7 @@ from __future__ import annotations from pathlib import Path # noqa: TCH003 required for Sphinx -from typing import ( - TYPE_CHECKING, - Any, - Callable, - Dict, - List, - Optional, - Protocol, - runtime_checkable, -) +from typing import TYPE_CHECKING, Any, Callable, Protocol, runtime_checkable if TYPE_CHECKING: import os @@ -260,8 +251,8 @@ def my_step(step: StepRunner) -> None: def setup_dependent( self, - original_step: "StepRunner", - current_step: "StepRunner", + original_step: StepRunner, + current_step: StepRunner, ) -> None: """ Run some logic into a dependent step. @@ -356,7 +347,7 @@ def coverage(self, step: StepRunner) -> None: .. tip:: The :py:func:`dwas.predefined.coverage` step does roughly this. """ - def gather_artifacts(self, step: "StepRunner") -> Dict[str, List[Any]]: + def gather_artifacts(self, step: StepRunner) -> dict[str, list[Any]]: """ Gather all artifacts exposed by this step. @@ -433,7 +424,7 @@ class StepRunner: standardized environment. """ - def __init__(self, handler: "StepHandler") -> None: + def __init__(self, handler: StepHandler) -> None: self._handler = handler @property @@ -483,7 +474,7 @@ def cache_path(self) -> Path: return self.config.cache_path / "cache" / name - def get_artifacts(self, key: str) -> List[Any]: + def get_artifacts(self, key: str) -> list[Any]: """ Get the artifacts exported by previous steps for the given key. @@ -520,12 +511,10 @@ def install(self, *packages: str) -> None: def run( self, - command: List[str], + command: list[str], *, - cwd: Optional[ - str | bytes | os.PathLike[str] | os.PathLike[bytes] - ] = None, - env: Optional[Dict[str, str]] = None, + cwd: str | bytes | os.PathLike[str] | os.PathLike[bytes] | None = None, + env: dict[str, str] | None = None, external_command: bool = False, silent_on_success: bool = False, ) -> subprocess.CompletedProcess[None]: diff --git a/src/dwas/_subproc.py b/src/dwas/_subproc.py index 5461226..42f9668 100644 --- a/src/dwas/_subproc.py +++ b/src/dwas/_subproc.py @@ -9,7 +9,7 @@ from contextlib import suppress from contextvars import copy_context from threading import Lock, Thread -from typing import Any, Dict, List, Optional, Set, TextIO +from typing import Any, TextIO from . import _io @@ -24,7 +24,7 @@ def _stream(source: int, dest: TextIO) -> None: class ProcessManager: def __init__(self) -> None: - self.processes: Set[subprocess.Popen[Any]] = set() + self.processes: set[subprocess.Popen[Any]] = set() self._lock = Lock() self._was_killed = False @@ -72,11 +72,9 @@ def kill(self) -> None: def run( self, - command: List[str], - env: Dict[str, str], - cwd: Optional[ - str | bytes | os.PathLike[str] | os.PathLike[bytes] - ] = None, + command: list[str], + env: dict[str, str], + cwd: str | bytes | os.PathLike[str] | os.PathLike[bytes] | None = None, *, silent_on_success: bool = False, ) -> subprocess.CompletedProcess[None]: diff --git a/src/dwas/predefined/_black.py b/src/dwas/predefined/_black.py index 2f3658d..51591ca 100644 --- a/src/dwas/predefined/_black.py +++ b/src/dwas/predefined/_black.py @@ -1,4 +1,6 @@ -from typing import List, Optional, Sequence +from __future__ import annotations + +from typing import Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -20,7 +22,7 @@ def __call__( self, step: StepRunner, files: Sequence[str], - additional_arguments: List[str], + additional_arguments: list[str], ) -> None: additional_arguments = additional_arguments.copy() @@ -36,8 +38,8 @@ def __call__( def black( *, - files: Optional[Sequence[str]] = None, - additional_arguments: Optional[List[str]] = None, + files: Sequence[str] | None = None, + additional_arguments: list[str] | None = None, ) -> Step: """ Run `the Black formatter`_ against your python source code. diff --git a/src/dwas/predefined/_coverage.py b/src/dwas/predefined/_coverage.py index adcbb56..cb72277 100644 --- a/src/dwas/predefined/_coverage.py +++ b/src/dwas/predefined/_coverage.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from __future__ import annotations # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -20,7 +20,7 @@ def __init__(self) -> None: def __call__( self, step: StepRunner, - reports: List[List[str]], + reports: list[list[str]], ) -> None: env = {"COVERAGE_FILE": str(step.cache_path / "coverage")} @@ -35,7 +35,7 @@ def __call__( step.run(["coverage", *report], env=env) -def coverage(*, reports: Optional[List[List[str]]] = None) -> Step: +def coverage(*, reports: list[list[str]] | None = None) -> Step: """ Run `coverage.py`_ to generate coverage reports. diff --git a/src/dwas/predefined/_docformatter.py b/src/dwas/predefined/_docformatter.py index fd47711..d786e39 100644 --- a/src/dwas/predefined/_docformatter.py +++ b/src/dwas/predefined/_docformatter.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import logging import subprocess -from typing import List, Optional, Sequence +from typing import Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -41,9 +43,9 @@ def __call__( def docformatter( *, - files: Optional[Sequence[str]] = None, - additional_arguments: Optional[List[str]] = None, - expected_status_codes: Optional[List[int]] = None, + files: Sequence[str] | None = None, + additional_arguments: list[str] | None = None, + expected_status_codes: list[int] | None = None, ) -> Step: """ Run `docformatter`_ against your python source code. diff --git a/src/dwas/predefined/_isort.py b/src/dwas/predefined/_isort.py index 295a754..cfba62e 100644 --- a/src/dwas/predefined/_isort.py +++ b/src/dwas/predefined/_isort.py @@ -1,4 +1,6 @@ -from typing import List, Optional, Sequence +from __future__ import annotations + +from typing import Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -20,7 +22,7 @@ def __call__( self, step: StepRunner, files: Sequence[str], - additional_arguments: List[str], + additional_arguments: list[str], ) -> None: additional_arguments = additional_arguments.copy() @@ -32,8 +34,8 @@ def __call__( def isort( *, - files: Optional[Sequence[str]] = None, - additional_arguments: Optional[List[str]] = None, + files: Sequence[str] | None = None, + additional_arguments: list[str] | None = None, ) -> Step: """ Run `the isort formatter`_ against your python source code. diff --git a/src/dwas/predefined/_mypy.py b/src/dwas/predefined/_mypy.py index b34f827..98447ec 100644 --- a/src/dwas/predefined/_mypy.py +++ b/src/dwas/predefined/_mypy.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import logging import os -from typing import List, Optional, Sequence +from typing import Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -20,7 +22,7 @@ def __call__( self, step: StepRunner, files: Sequence[str], - additional_arguments: List[str], + additional_arguments: list[str], ) -> None: env = {} if step.config.colors: @@ -40,8 +42,8 @@ def __call__( def mypy( *, - files: Optional[Sequence[str]] = None, - additional_arguments: Optional[List[str]] = None, + files: Sequence[str] | None = None, + additional_arguments: list[str] | None = None, ) -> Step: """ Run `mypy`_ against your python source code. diff --git a/src/dwas/predefined/_package.py b/src/dwas/predefined/_package.py index c4eb9c7..7d3d0d4 100644 --- a/src/dwas/predefined/_package.py +++ b/src/dwas/predefined/_package.py @@ -1,7 +1,9 @@ +from __future__ import annotations + import logging import shutil from contextlib import suppress -from typing import Any, Dict, List +from typing import Any # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -21,7 +23,7 @@ class Package(StepWithDependentSetup): def __init__(self) -> None: self.__name__ = "package" - def gather_artifacts(self, step: StepRunner) -> Dict[str, List[Any]]: + def gather_artifacts(self, step: StepRunner) -> dict[str, list[Any]]: artifacts = {} sdists = [str(p) for p in step.cache_path.glob("*.tar.gz")] if sdists: diff --git a/src/dwas/predefined/_pylint.py b/src/dwas/predefined/_pylint.py index 8aa5baf..aacd384 100644 --- a/src/dwas/predefined/_pylint.py +++ b/src/dwas/predefined/_pylint.py @@ -1,4 +1,6 @@ -from typing import List, Optional, Sequence +from __future__ import annotations + +from typing import Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -20,7 +22,7 @@ def __call__( self, step: StepRunner, files: Sequence[str], - additional_arguments: List[str], + additional_arguments: list[str], ) -> None: additional_arguments = additional_arguments.copy() @@ -37,8 +39,8 @@ def __call__( def pylint( *, - files: Optional[Sequence[str]] = None, - additional_arguments: Optional[Sequence[str]] = None, + files: Sequence[str] | None = None, + additional_arguments: Sequence[str] | None = None, ) -> Step: """ Run `pylint`_ against your source code. diff --git a/src/dwas/predefined/_pytest.py b/src/dwas/predefined/_pytest.py index b87cf85..00e697f 100644 --- a/src/dwas/predefined/_pytest.py +++ b/src/dwas/predefined/_pytest.py @@ -1,11 +1,15 @@ +from __future__ import annotations + import logging -from pathlib import Path -from typing import Any, Dict, List, Optional, Sequence +from typing import TYPE_CHECKING, Any, Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it from .. import Step, StepRunner, parametrize, set_defaults +if TYPE_CHECKING: + from pathlib import Path + LOGGER = logging.getLogger(__name__) @@ -14,7 +18,7 @@ class Pytest(Step): def __init__(self) -> None: self.__name__ = "pytest" - def gather_artifacts(self, step: StepRunner) -> Dict[str, List[Any]]: + def gather_artifacts(self, step: StepRunner) -> dict[str, list[Any]]: coverage_file = self._get_coverage_file(step) if coverage_file is None or not coverage_file.exists(): return {} @@ -25,7 +29,7 @@ def __call__( self, step: StepRunner, args: Sequence[str], - user_args: Optional[Sequence[str]], + user_args: Sequence[str] | None, ) -> None: if user_args is None: user_args = [] @@ -39,7 +43,7 @@ def _get_coverage_file(self, step: StepRunner) -> Path: return step.cache_path / "reports" / "coverage" -def pytest(*, args: Optional[Sequence[str]] = None) -> Step: +def pytest(*, args: Sequence[str] | None = None) -> Step: """ Run `pytest`_. diff --git a/src/dwas/predefined/_ruff.py b/src/dwas/predefined/_ruff.py index 066d3b7..a676513 100644 --- a/src/dwas/predefined/_ruff.py +++ b/src/dwas/predefined/_ruff.py @@ -1,4 +1,6 @@ -from typing import List, Optional, Sequence +from __future__ import annotations + +from typing import Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -20,7 +22,7 @@ def __call__( self, step: StepRunner, files: Sequence[str], - additional_arguments: List[str], + additional_arguments: list[str], ) -> None: step.run( ["ruff", *additional_arguments, *files], @@ -30,8 +32,8 @@ def __call__( def ruff( *, - files: Optional[Sequence[str]] = None, - additional_arguments: Optional[List[str]] = None, + files: Sequence[str] | None = None, + additional_arguments: list[str] | None = None, ) -> Step: """ Run `Ruff`_ against your python source code. diff --git a/src/dwas/predefined/_sphinx.py b/src/dwas/predefined/_sphinx.py index 71c4cd9..9e414fc 100644 --- a/src/dwas/predefined/_sphinx.py +++ b/src/dwas/predefined/_sphinx.py @@ -1,7 +1,8 @@ +from __future__ import annotations + import shutil from contextlib import suppress -from pathlib import Path -from typing import Optional, Union +from pathlib import Path # noqa: TCH003 required for sphinx documentation from .. import Step, StepRunner, build_parameters, set_defaults @@ -19,7 +20,7 @@ class Sphinx(Step): def __init__(self) -> None: self.__name__ = "sphinx" - def clean(self, output: Optional[Union[Path, str]]) -> None: + def clean(self, output: Path | str | None) -> None: if output is not None: with suppress(FileNotFoundError): shutil.rmtree(output) @@ -28,8 +29,8 @@ def __call__( self, step: StepRunner, builder: str, - sourcedir: Union[Path, str], - output: Optional[Union[Path, str]], + sourcedir: Path | str, + output: Path | str | None, *, warning_as_error: bool, ) -> None: @@ -63,10 +64,10 @@ def __call__( def sphinx( *, - builder: Optional[str] = None, - sourcedir: Optional[Union[Path, str]] = None, - output: Optional[Union[Path, str]] = None, - warning_as_error: Optional[bool] = None, + builder: str | None = None, + sourcedir: Path | str | None = None, + output: Path | str | None = None, + warning_as_error: bool | None = None, ) -> Step: """ Run `sphinx`_. diff --git a/src/dwas/predefined/_twine.py b/src/dwas/predefined/_twine.py index 7b500fe..58b84f4 100644 --- a/src/dwas/predefined/_twine.py +++ b/src/dwas/predefined/_twine.py @@ -1,5 +1,6 @@ +from __future__ import annotations + import logging -from typing import List, Optional # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -19,7 +20,7 @@ def __init__(self) -> None: self.__name__ = "twine" def __call__( - self, step: StepRunner, additional_arguments: List[str] + self, step: StepRunner, additional_arguments: list[str] ) -> None: sdists = step.get_artifacts("sdists") wheels = step.get_artifacts("wheels") @@ -36,7 +37,7 @@ def __call__( step.run(["twine", *additional_arguments, *sdists, *wheels]) -def twine(*, additional_arguments: Optional[List[str]] = None) -> Step: +def twine(*, additional_arguments: list[str] | None = None) -> Step: """ Run `twine`_ against the provided packages. diff --git a/src/dwas/predefined/_unimport.py b/src/dwas/predefined/_unimport.py index e5d2322..3803a42 100644 --- a/src/dwas/predefined/_unimport.py +++ b/src/dwas/predefined/_unimport.py @@ -1,4 +1,6 @@ -from typing import List, Optional, Sequence +from __future__ import annotations + +from typing import Sequence # XXX: All imports here should be done from the top level. If we need it, # users might need it @@ -20,7 +22,7 @@ def __call__( self, step: StepRunner, files: Sequence[str], - additional_arguments: List[str], + additional_arguments: list[str], ) -> None: additional_arguments = additional_arguments.copy() @@ -35,8 +37,8 @@ def __call__( def unimport( *, - files: Optional[Sequence[str]] = None, - additional_arguments: Optional[List[str]] = None, + files: Sequence[str] | None = None, + additional_arguments: list[str] | None = None, ) -> Step: """ Run `the Unimport formatter`_ against your python source code. diff --git a/tests/_utils.py b/tests/_utils.py index 97390be..f8c91b7 100644 --- a/tests/_utils.py +++ b/tests/_utils.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import functools import logging import re @@ -5,8 +7,7 @@ from contextlib import contextmanager from contextvars import Context from dataclasses import dataclass -from pathlib import Path -from typing import Any, Callable, Iterator, List, Optional, TypeVar, Union +from typing import TYPE_CHECKING, Any, Callable, Iterator, TypeVar import pytest from _pytest.capture import FDCapture, MultiCapture @@ -14,6 +15,9 @@ from dwas.__main__ import main from tests import TESTS_PATH +if TYPE_CHECKING: + from pathlib import Path + _T = TypeVar("_T") ANSI_COLOR_CODES_RE = re.compile(r"\x1B\[\dm") @@ -58,13 +62,13 @@ def wrapper(func): @dataclass(frozen=True) class Result: - exc: Optional[SystemExit] + exc: SystemExit | None stdout: str stderr: str @isolated_context -def execute(args: List[str], expected_status: int = 0) -> Result: +def execute(args: list[str], expected_status: int = 0) -> Result: """ Run dwas in an isolated context and returns the result from the run. @@ -76,7 +80,7 @@ def execute(args: List[str], expected_status: int = 0) -> Result: exception = None # See https://github.com/python/typeshed/issues/8513#issue-1333671093 - exit_code: Union[str, int, None] = 0 + exit_code: str | int | None = 0 try: with isolated_logging(): @@ -100,10 +104,10 @@ def execute(args: List[str], expected_status: int = 0) -> Result: def cli( *, - steps: Optional[List[str]] = None, + steps: list[str] | None = None, cache_path: Path, - colors: Optional[bool] = None, - except_steps: Optional[List[str]] = None, + colors: bool | None = None, + except_steps: list[str] | None = None, expected_status: int = 0, ) -> Result: args = ["--verbose"] diff --git a/tests/test_registration.py b/tests/test_registration.py index b296cf4..a948b1f 100644 --- a/tests/test_registration.py +++ b/tests/test_registration.py @@ -1,6 +1,8 @@ # This tests some internals # ruff: noqa:SLF001 -from typing import Any, Dict, Optional +from __future__ import annotations + +from typing import Any import pytest @@ -13,8 +15,8 @@ from ._utils import isolated_context -def _get_all_steps_from_pipeline(pipeline: Pipeline) -> Dict[str, Any]: - def _format_step(step: BaseStepHandler) -> Dict[str, Any]: +def _get_all_steps_from_pipeline(pipeline: Pipeline) -> dict[str, Any]: + def _format_step(step: BaseStepHandler) -> dict[str, Any]: if isinstance(step, StepGroupHandler): return { "type": "group", @@ -45,10 +47,10 @@ def _format_step(step: BaseStepHandler) -> Dict[str, Any]: def _expect_step( *, - python: Optional[str] = None, - run_by_default: Optional[bool] = None, - parameters: Dict[str, Any], -) -> Dict[str, Any]: + python: str | None = None, + run_by_default: bool | None = None, + parameters: dict[str, Any], +) -> dict[str, Any]: if run_by_default is None: run_by_default = True @@ -145,7 +147,7 @@ def noop(): if from_parameters: noop = parametrize("dependencies", [["one"]])(noop) - kwargs: Dict[str, Any] = {} + kwargs: dict[str, Any] = {} else: kwargs = {"dependencies": ["one"]} diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index f4329a0..007d3fa 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -1,4 +1,5 @@ -from typing import Dict, List, Optional, Set +from __future__ import annotations + from unittest.mock import ANY import pytest @@ -9,14 +10,14 @@ def assert_scheduler_state( scheduler: Scheduler, - waiting: Optional[Set[str]] = None, - ready: Optional[List[str]] = None, - running: Optional[Dict[str, float]] = None, - done: Optional[Set[str]] = None, - failed: Optional[Set[str]] = None, - blocked: Optional[Set[str]] = None, - cancelled: Optional[Set[str]] = None, - skipped: Optional[Set[str]] = None, + waiting: set[str] | None = None, + ready: list[str] | None = None, + running: dict[str, float] | None = None, + done: set[str] | None = None, + failed: set[str] | None = None, + blocked: set[str] | None = None, + cancelled: set[str] | None = None, + skipped: set[str] | None = None, ) -> None: assert { "waiting": scheduler.waiting,