Skip to content

kraken.std.python

kraken.std.python

BlackTask

Bases: EnvironmentAwareDispatchTask

A task to run the black formatter to either check for necessary changes or apply changes.

Source code in kraken/std/python/tasks/black_task.py 
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
class BlackTask(EnvironmentAwareDispatchTask):
    """A task to run the `black` formatter to either check for necessary changes or apply changes."""

    python_dependencies = ["black"]

    black_bin: Property[str] = Property.default("black")
    check_only: Property[bool] = Property.default(False)
    config_file: Property[Path]
    additional_args: Property[Sequence[str]] = Property.default_factory(list)
    additional_files: Property[Sequence[Path]] = Property.default_factory(list)

    # EnvironmentAwareDispatchTask

    def get_execute_command(self) -> list[str]:
        command = [self.black_bin.get(), str(self.settings.source_directory)]
        command += self.settings.get_tests_directory_as_args()
        command += [str(directory) for directory in self.settings.lint_enforced_directories]
        command += [str(p) for p in self.additional_files.get()]
        if self.check_only.get():
            command += ["--check", "--diff"]
        if self.config_file.is_filled():
            command += ["--config", str(self.config_file.get().absolute())]
        command += self.additional_args.get()
        return command

    # Task

    def get_description(self) -> str | None:
        if self.check_only.get():
            return "Check Python source files formatting with Black."
        else:
            return "Format Python source files with Black."

Flake8Task

Bases: EnvironmentAwareDispatchTask

Lint Python source files with Flake8.

Source code in kraken/std/python/tasks/flake8_task.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class Flake8Task(EnvironmentAwareDispatchTask):
    """
    Lint Python source files with Flake8.
    """

    description = "Lint Python source files with Flake8."
    python_dependencies = ["flake8"]

    flake8_bin: Property[str] = Property.default("flake8")
    config_file: Property[Path]
    additional_args: Property[list[str]] = Property.default_factory(list)

    # EnvironmentAwareDispatchTask

    def get_execute_command(self) -> list[str]:
        command = [
            self.flake8_bin.get(),
            str(self.settings.source_directory),
        ] + self.settings.get_tests_directory_as_args()
        command += [str(directory) for directory in self.settings.lint_enforced_directories]
        if self.config_file.is_filled():
            command += ["--config", str(self.config_file.get().absolute())]
        command += self.additional_args.get()
        return command

InfoTask

Bases: Task

Source code in kraken/std/python/tasks/info_task.py 
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
class InfoTask(Task):
    build_system: Property[PythonBuildSystem]

    def get_description(self) -> str:
        return "Displays metadata about the Kraken-managed environment."

    def execute(self) -> TaskStatus:
        """Displays metadata about the Kraken-managed environment."""
        python_path = self.get_python_path()
        virtual_env_path = self.get_virtual_env_path()
        try:
            version = self.get_python_version()
        except subprocess.CalledProcessError as error:
            return TaskStatus.failed(f"Error while getting the version of the current Python interpreter: {error}")

        print(
            colored(f" ---------- {self.build_system.get().name}-managed environment information ----------", "magenta")
        )
        print(colored("Python version:           ", "cyan"), colored(f"{version.strip()}", "blue"))
        print(colored("Python path:              ", "cyan"), colored(f"{python_path}", "blue"))
        print(colored("Virtual environment path: ", "cyan"), colored(f"{virtual_env_path}", "blue"))
        print(colored(" ------------------------------------------------------------", "magenta"))

        return TaskStatus.succeeded()

    def get_virtual_env_path(self) -> Path:
        """
        Returns the current virtual environment path
        """
        return self.build_system.get().get_managed_environment().get_path()

    def get_python_path(self) -> Path:
        """Returns the path of the Python interpreter of the Kraken-managed environment."""
        return self.get_virtual_env_path() / "bin" / "python"

    def get_python_version(self) -> str:
        """Returns the version of the Python interpreter of the Kraken-managed environment."""
        return subprocess.run(
            [self.get_python_path(), "--version"], stdout=subprocess.PIPE, shell=False, check=True
        ).stdout.decode("utf-8")
execute 
execute() -> TaskStatus

Displays metadata about the Kraken-managed environment.

Source code in kraken/std/python/tasks/info_task.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def execute(self) -> TaskStatus:
    """Displays metadata about the Kraken-managed environment."""
    python_path = self.get_python_path()
    virtual_env_path = self.get_virtual_env_path()
    try:
        version = self.get_python_version()
    except subprocess.CalledProcessError as error:
        return TaskStatus.failed(f"Error while getting the version of the current Python interpreter: {error}")

    print(
        colored(f" ---------- {self.build_system.get().name}-managed environment information ----------", "magenta")
    )
    print(colored("Python version:           ", "cyan"), colored(f"{version.strip()}", "blue"))
    print(colored("Python path:              ", "cyan"), colored(f"{python_path}", "blue"))
    print(colored("Virtual environment path: ", "cyan"), colored(f"{virtual_env_path}", "blue"))
    print(colored(" ------------------------------------------------------------", "magenta"))

    return TaskStatus.succeeded()
get_python_path 
get_python_path() -> Path

Returns the path of the Python interpreter of the Kraken-managed environment.

Source code in kraken/std/python/tasks/info_task.py 
43
44
45
def get_python_path(self) -> Path:
    """Returns the path of the Python interpreter of the Kraken-managed environment."""
    return self.get_virtual_env_path() / "bin" / "python"
get_python_version 
get_python_version() -> str

Returns the version of the Python interpreter of the Kraken-managed environment.

Source code in kraken/std/python/tasks/info_task.py 
47
48
49
50
51
def get_python_version(self) -> str:
    """Returns the version of the Python interpreter of the Kraken-managed environment."""
    return subprocess.run(
        [self.get_python_path(), "--version"], stdout=subprocess.PIPE, shell=False, check=True
    ).stdout.decode("utf-8")
get_virtual_env_path 
get_virtual_env_path() -> Path

Returns the current virtual environment path

Source code in kraken/std/python/tasks/info_task.py 
37
38
39
40
41
def get_virtual_env_path(self) -> Path:
    """
    Returns the current virtual environment path
    """
    return self.build_system.get().get_managed_environment().get_path()

PublishTask

Bases: Task

Publishes Python distributions to one or more indexes using :mod:twine.

Source code in kraken/std/python/tasks/publish_task.py 
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
60
61
class PublishTask(Task):
    """Publishes Python distributions to one or more indexes using :mod:`twine`."""

    description = "Upload the distributions of your Python project. [index url: %(index_upload_url)s]"
    twine_bin: Property[Path]
    index_upload_url: Property[str]
    index_credentials: Property[tuple[str, str] | None] = Property.default(None)
    distributions: Property[list[Path]]
    skip_existing: Property[bool] = Property.default(False)
    interactive: Property[bool] = Property.default(True)
    dependencies: list[Task]

    def __init__(self, name: str, project: Project) -> None:
        super().__init__(name, project)
        self.dependencies = []

    def get_relationships(self) -> Iterable[TaskRelationship]:
        yield from (TaskRelationship(task, True, False) for task in self.dependencies)
        yield from super().get_relationships()

    def execute(self) -> TaskStatus:
        credentials = self.index_credentials.get()
        repository_url = self.index_upload_url.get().rstrip("/") + "/"
        command = [
            str(self.twine_bin.get()),
            "upload",
            "--repository-url",
            repository_url,
            "--verbose",
            *[str(x.absolute()) for x in self.distributions.get()],
        ]
        if credentials:
            command += [
                "--username",
                credentials[0],
                "--password",
                credentials[1],
            ]
        if not self.interactive.get():
            command.append("--non-interactive")
        if self.skip_existing.get():
            command.append("--skip-existing")

        safe_command = [x.replace(credentials[1], "MASKED") for x in command] if credentials else command
        self.logger.info("$ %s", safe_command)

        returncode = subprocess.call(command, cwd=self.project.directory)
        return TaskStatus.from_exit_code(safe_command, returncode)

PyclnTask

Bases: EnvironmentAwareDispatchTask

A task to run the pycln formatter to either check for necessary changes or apply changes.

Source code in kraken/std/python/tasks/pycln_task.py 
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
class PyclnTask(EnvironmentAwareDispatchTask):
    """A task to run the `pycln` formatter to either check for necessary changes or apply changes."""

    python_dependencies = ["pycln"]

    pycln_bin: Property[str] = Property.default("pycln")
    check_only: Property[bool] = Property.default(False)
    config_file: Property[Path]
    additional_args: Property[list[str]] = Property.default_factory(list)
    additional_files: Property[list[Path]] = Property.default_factory(list)

    # EnvironmentAwareDispatchTask

    def get_execute_command(self) -> list[str]:
        command = [self.pycln_bin.get(), str(self.settings.source_directory)]
        command += self.settings.get_tests_directory_as_args()
        command += [str(directory) for directory in self.settings.lint_enforced_directories]
        command += [str(p) for p in self.additional_files.get()]
        if self.check_only.get():
            command += ["--check", "--diff"]
        if self.config_file.is_filled():
            command += ["--config", str(self.config_file.get().absolute())]
        command += self.additional_args.get()
        return command

    def get_description(self) -> str:
        if self.check_only.get():
            return "Check Python imports with Pycln."
        else:
            return "Remove unused Python imports with Pycln."

PythonSettings dataclass

Project-global settings for Python tasks.

Source code in kraken/std/python/settings.py 
 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
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@dataclass
class PythonSettings:
    """Project-global settings for Python tasks."""

    @dataclass
    class _PackageIndex(PackageIndex):
        """
        Extends the #PackageIndex with additional fields we need for Python package indexes at runtime in Kraken.
        """

        #: An alternative URL to upload packages to.
        upload_url: str | None

        #: Credentials to use when publishing to or reading from the index.
        credentials: tuple[str, str] | None

        #: Whether this index should be used to source packages from.
        is_package_source: bool

        #: Whether this index should be used to publish packages to.
        publish: bool

    project: Project
    build_system: PythonBuildSystem | None = None
    source_directory: Path = Path("src")
    tests_directory: Path | None = None
    lint_enforced_directories: list[Path] = field(default_factory=list)
    package_indexes: dict[str, _PackageIndex] = field(default_factory=dict)
    always_use_managed_env: bool = True
    skip_install_if_venv_exists: bool = True

    def get_tests_directory(self) -> Path | None:
        """Returns :attr:`tests_directory` if it is set. If not, it will look for the following directories and
        return the first that exists: `test/`, `tests/`, `src/test/`, `src/tests/`. The determined path will be
        relative to the project directory."""

        if self.tests_directory and self.tests_directory.is_dir():
            return self.tests_directory
        for test_dir in map(Path, ["test", "tests", "src/test", "src/tests"]):
            if (self.project.directory / test_dir).is_dir():
                return test_dir
        return None

    def get_tests_directory_as_args(self) -> list[str]:
        """Returns a list with a single item that is the test directory, or an empty list. This is convenient
        when constructing command-line arguments where you want to pass the test directory if it exists."""

        test_dir = self.get_tests_directory()
        return [] if test_dir is None else [str(test_dir)]

    def get_default_package_index(self) -> _PackageIndex | None:
        return next(
            (
                index
                for index in self.package_indexes.values()
                if index.priority.value == PackageIndex.Priority.default.value
            ),
            None,
        )

    def add_package_index(
        self,
        alias: str,
        *,
        index_url: str | None = None,
        upload_url: str | None = None,
        credentials: tuple[str, str] | None = None,
        is_package_source: bool = True,
        priority: PackageIndex.Priority | str = PackageIndex.Priority.supplemental,
        publish: bool = False,
    ) -> PythonSettings:
        """Adds an index to consume Python packages from or publish packages to.

        :param alias: An alias for the package index.
        :param index_url: The URL of the package index (with the trailing `/simple` bit if applicable).
            If not specified, *alias* must be a known package index (`pypi` or `testpypi`).
        :param upload_url: If the upload url deviates from the registry URL. Otherwise, the upload URL will
            be the same as the *url*,
        :param credentials: Optional credentials to read from the index.
        :param is_package_source: If set to `False`, the index will not be used to source packages from, but
            can be used to publish to.
        :param publish: Whether publishing to this index should be enabled.
        """

        if isinstance(priority, str):
            priority = PackageIndex.Priority[priority]

        if priority == PackageIndex.Priority.default:
            defidx = self.get_default_package_index()
            if defidx is not None and defidx.alias != alias:
                raise ValueError(f"cannot add another default index (got: {defidx.alias!r}, trying to add: {alias!r})")

        if index_url is None:
            if alias == "pypi":
                index_url = "https://pypi.org/simple"
            elif alias == "testpypi":
                index_url = "https://test.pypi.org/simple"
            else:
                raise ValueError(f"cannot derive index URL for alias {alias!r}")
        if upload_url is None:
            if alias == "pypi":
                upload_url = "https://upload.pypi.org/legacy"
            elif alias == "testpypi":
                upload_url = "https://test.pypi.org/legacy"
            elif index_url.endswith("/simple"):
                upload_url = index_url[: -len("/simple")]
            else:
                raise ValueError(f"cannot derive upload URL for alias {alias!r} and index URL {index_url!r}")

        self.package_indexes[alias] = self._PackageIndex(
            alias=alias,
            index_url=index_url,
            priority=priority,
            upload_url=upload_url,
            credentials=credentials,
            is_package_source=is_package_source,
            publish=publish,
            verify_ssl=True,
        )
        return self

    def get_primary_index(self) -> _PackageIndex | None:
        default: PythonSettings._PackageIndex | None = None
        for idx in self.package_indexes.values():
            if idx.priority == PackageIndex.Priority.primary:
                return idx
            if idx.priority == PackageIndex.Priority.default:
                default = idx
        return default
add_package_index 
add_package_index(
    alias: str,
    *,
    index_url: str | None = None,
    upload_url: str | None = None,
    credentials: tuple[str, str] | None = None,
    is_package_source: bool = True,
    priority: (
        Priority | str
    ) = PackageIndex.Priority.supplemental,
    publish: bool = False
) -> PythonSettings

Adds an index to consume Python packages from or publish packages to.

:param alias: An alias for the package index. :param index_url: The URL of the package index (with the trailing /simple bit if applicable). If not specified, alias must be a known package index (pypi or testpypi). :param upload_url: If the upload url deviates from the registry URL. Otherwise, the upload URL will be the same as the url, :param credentials: Optional credentials to read from the index. :param is_package_source: If set to False, the index will not be used to source packages from, but can be used to publish to. :param publish: Whether publishing to this index should be enabled.

Source code in kraken/std/python/settings.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def add_package_index(
    self,
    alias: str,
    *,
    index_url: str | None = None,
    upload_url: str | None = None,
    credentials: tuple[str, str] | None = None,
    is_package_source: bool = True,
    priority: PackageIndex.Priority | str = PackageIndex.Priority.supplemental,
    publish: bool = False,
) -> PythonSettings:
    """Adds an index to consume Python packages from or publish packages to.

    :param alias: An alias for the package index.
    :param index_url: The URL of the package index (with the trailing `/simple` bit if applicable).
        If not specified, *alias* must be a known package index (`pypi` or `testpypi`).
    :param upload_url: If the upload url deviates from the registry URL. Otherwise, the upload URL will
        be the same as the *url*,
    :param credentials: Optional credentials to read from the index.
    :param is_package_source: If set to `False`, the index will not be used to source packages from, but
        can be used to publish to.
    :param publish: Whether publishing to this index should be enabled.
    """

    if isinstance(priority, str):
        priority = PackageIndex.Priority[priority]

    if priority == PackageIndex.Priority.default:
        defidx = self.get_default_package_index()
        if defidx is not None and defidx.alias != alias:
            raise ValueError(f"cannot add another default index (got: {defidx.alias!r}, trying to add: {alias!r})")

    if index_url is None:
        if alias == "pypi":
            index_url = "https://pypi.org/simple"
        elif alias == "testpypi":
            index_url = "https://test.pypi.org/simple"
        else:
            raise ValueError(f"cannot derive index URL for alias {alias!r}")
    if upload_url is None:
        if alias == "pypi":
            upload_url = "https://upload.pypi.org/legacy"
        elif alias == "testpypi":
            upload_url = "https://test.pypi.org/legacy"
        elif index_url.endswith("/simple"):
            upload_url = index_url[: -len("/simple")]
        else:
            raise ValueError(f"cannot derive upload URL for alias {alias!r} and index URL {index_url!r}")

    self.package_indexes[alias] = self._PackageIndex(
        alias=alias,
        index_url=index_url,
        priority=priority,
        upload_url=upload_url,
        credentials=credentials,
        is_package_source=is_package_source,
        publish=publish,
        verify_ssl=True,
    )
    return self
get_tests_directory 
get_tests_directory() -> Path | None

Returns :attr:tests_directory if it is set. If not, it will look for the following directories and return the first that exists: test/, tests/, src/test/, src/tests/. The determined path will be relative to the project directory.

Source code in kraken/std/python/settings.py 
46
47
48
49
50
51
52
53
54
55
56
def get_tests_directory(self) -> Path | None:
    """Returns :attr:`tests_directory` if it is set. If not, it will look for the following directories and
    return the first that exists: `test/`, `tests/`, `src/test/`, `src/tests/`. The determined path will be
    relative to the project directory."""

    if self.tests_directory and self.tests_directory.is_dir():
        return self.tests_directory
    for test_dir in map(Path, ["test", "tests", "src/test", "src/tests"]):
        if (self.project.directory / test_dir).is_dir():
            return test_dir
    return None
get_tests_directory_as_args 
get_tests_directory_as_args() -> list[str]

Returns a list with a single item that is the test directory, or an empty list. This is convenient when constructing command-line arguments where you want to pass the test directory if it exists.

Source code in kraken/std/python/settings.py 
58
59
60
61
62
63
def get_tests_directory_as_args(self) -> list[str]:
    """Returns a list with a single item that is the test directory, or an empty list. This is convenient
    when constructing command-line arguments where you want to pass the test directory if it exists."""

    test_dir = self.get_tests_directory()
    return [] if test_dir is None else [str(test_dir)]

black 

black(
    *,
    name: str = "python.black",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_args: (
        Sequence[str] | Supplier[Sequence[str]]
    ) = (),
    additional_files: (
        Sequence[Path] | Supplier[Sequence[Path]]
    ) = (),
    version_spec: str | None = None
) -> BlackTasks

Creates two black tasks, one to check and another to format. The check task will be grouped under "lint" whereas the format task will be grouped under "fmt".

:param version_spec: If specified, the Black tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/black_task.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def black(
    *,
    name: str = "python.black",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_args: Sequence[str] | Supplier[Sequence[str]] = (),
    additional_files: Sequence[Path] | Supplier[Sequence[Path]] = (),
    version_spec: str | None = None,
) -> BlackTasks:
    """Creates two black tasks, one to check and another to format. The check task will be grouped under `"lint"`
    whereas the format task will be grouped under `"fmt"`.

    :param version_spec: If specified, the Black tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    project = project or Project.current()

    if version_spec is not None:
        black_bin = pex_build(
            "black", requirements=[f"black{version_spec}"], console_script="black", project=project
        ).output_file.map(str)
    else:
        black_bin = Supplier.of("black")

    check_task = project.task(f"{name}.check", BlackTask, group="lint")
    check_task.black_bin = black_bin
    check_task.check_only = True
    check_task.config_file = config_file
    check_task.additional_args = additional_args
    check_task.additional_files = additional_files

    format_task = project.task(name, BlackTask, group="fmt")
    format_task.black_bin = black_bin
    format_task.check_only = False
    format_task.config_file = config_file
    format_task.additional_args = additional_args
    format_task.additional_files = additional_files

    return BlackTasks(check_task, format_task)

build 

build(
    *,
    name: str = "python.build",
    group: str | None = "build",
    as_version: str | None = None,
    project: Project | None = None
) -> BuildTask

Creates a build task for the given project.

The build task relies on the build system configured in the Python project settings.

Source code in kraken/std/python/tasks/build_task.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def build(
    *,
    name: str = "python.build",
    group: str | None = "build",
    as_version: str | None = None,
    project: Project | None = None,
) -> BuildTask:
    """Creates a build task for the given project.

    The build task relies on the build system configured in the Python project settings."""

    project = project or Project.current()
    task = project.task(name, BuildTask, group=group)
    task.build_system = Supplier.of_callable(lambda: python_settings(project).build_system)
    task.as_version = as_version
    return task

flake8 

flake8(
    *,
    name: str = "python.flake8",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    version_spec: str | None = None,
    additional_requirements: Sequence[str] = ()
) -> Flake8Task

Creates a task for linting your Python project with Flake8.

:param version_spec: If specified, the Flake8 tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/flake8_task.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def flake8(
    *,
    name: str = "python.flake8",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    version_spec: str | None = None,
    additional_requirements: Sequence[str] = (),
) -> Flake8Task:
    """Creates a task for linting your Python project with Flake8.

    :param version_spec: If specified, the Flake8 tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    project = project or Project.current()

    if version_spec is not None:
        flake8_bin = pex_build(
            "flake8",
            requirements=[f"flake8{version_spec}", *additional_requirements],
            console_script="flake8",
            project=project,
        ).output_file.map(str)
    else:
        flake8_bin = Supplier.of("flake8")

    task = project.task(name, Flake8Task, group="lint")
    task.flake8_bin = flake8_bin
    if config_file is not None:
        task.config_file = config_file
    return task

git_version_to_python_version 

git_version_to_python_version(
    value: str | GitVersion, include_sha: bool
) -> str

Converts a Git version to a Python version.

:param value: The Git version to convert. :param sha: Include the SHA of the commit distance if it exists.

Source code in kraken/std/python/version.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def git_version_to_python_version(value: str | GitVersion, include_sha: bool) -> str:
    """Converts a Git version to a Python version.

    :param value: The Git version to convert.
    :param sha: Include the SHA of the commit distance if it exists.
    """

    version = GitVersion.parse(value) if isinstance(value, str) else value
    final_version = f"{version.major}.{version.minor}.{version.patch}"
    if version.pre_release:
        final_version = f"{final_version}{_PRE_RELEASE_NAMES[version.pre_release.kind]}{version.pre_release.value}"
    if version.distance:
        final_version += f".post0.dev{version.distance.value}"
        if include_sha:
            final_version += f"+g{version.distance.sha}"
            if version.dirty:
                final_version += "-dirty"
    if version.dirty and "-dirty" not in final_version:
        final_version += "+dirty"
    return final_version

info 

info(
    *,
    project: Project | None = None,
    build_system: PythonBuildSystem | None = None
) -> InfoTask

This task displays a list of useful info on current Python and virtual environment settings.

Source code in kraken/std/python/tasks/info_task.py 
54
55
56
57
58
59
60
61
62
63
64
65
def info(*, project: Project | None = None, build_system: PythonBuildSystem | None = None) -> InfoTask:
    """
    This task displays a list of useful info on current Python and virtual environment settings.
    """

    project = project or Project.current()
    if build_system is None:
        build_system = python_settings().build_system

    task = project.task("python.info", InfoTask)
    task.build_system = build_system
    return task

install 

install(
    *,
    name: str = "python.install",
    project: Project | None = None
) -> InstallTask

Get or create the python.install task for the given project.

The install task relies on the build system configured in the Python project settings.

Source code in kraken/std/python/tasks/install_task.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def install(*, name: str = "python.install", project: Project | None = None) -> InstallTask:
    """Get or create the `python.install` task for the given project.

    The install task relies on the build system configured in the Python project settings."""

    project = project or Project.current()
    task = cast(Union[InstallTask, None], project.tasks().get(name))
    if task is None:
        task = project.task(name, InstallTask)
        task.build_system = Supplier.of_callable(lambda: python_settings(project).build_system)
        task.always_use_managed_env = Supplier.of_callable(lambda: python_settings(project).always_use_managed_env)
        task.skip_if_venv_exists = Supplier.of_callable(lambda: python_settings(project).skip_install_if_venv_exists)

    return task

isort 

isort(
    *,
    name: str = "python.isort",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_files: (
        Sequence[Path] | Supplier[Sequence[Path]]
    ) = (),
    version_spec: str | None = None
) -> IsortTasks

:param version_spec: If specified, the isort tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/isort_task.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def isort(
    *,
    name: str = "python.isort",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_files: Sequence[Path] | Supplier[Sequence[Path]] = (),
    version_spec: str | None = None,
) -> IsortTasks:
    """
    :param version_spec: If specified, the isort tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    # TODO (@NiklasRosenstein): We may need to ensure an order to isort and block somehow, sometimes they yield
    #       slightly different results based on the order they run.
    project = project or Project.current()

    if version_spec is not None:
        isort_bin = pex_build(
            "isort", requirements=[f"isort{version_spec}"], console_script="isort", project=project
        ).output_file.map(str)
    else:
        isort_bin = Supplier.of("isort")

    check_task = project.task(f"{name}.check", IsortTask, group="lint")
    check_task.isort_bin = isort_bin
    check_task.check_only = True
    check_task.config_file = config_file
    check_task.additional_files = additional_files

    format_task = project.task(name, IsortTask, group="fmt")
    format_task.isort_bin = isort_bin
    format_task.check_only = False
    format_task.config_file = config_file
    format_task.additional_files = additional_files

    return IsortTasks(check_task, format_task)

mypy 

mypy(
    *,
    name: str = "python.mypy",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_args: (
        Sequence[str] | Supplier[Sequence[str]]
    ) = (),
    check_tests: bool = True,
    use_daemon: bool = True,
    python_version: str | Supplier[str] | None = None,
    version_spec: str | None = None
) -> MypyTask

:param version_spec: If specified, the Mypy tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/mypy_task.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def mypy(
    *,
    name: str = "python.mypy",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_args: Sequence[str] | Supplier[Sequence[str]] = (),
    check_tests: bool = True,
    use_daemon: bool = True,
    python_version: str | Supplier[str] | None = None,
    version_spec: str | None = None,
) -> MypyTask:
    """
    :param version_spec: If specified, the Mypy tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    project = project or Project.current()

    if version_spec is not None:
        mypy_pex_bin = pex_build(
            "mypy", requirements=[f"mypy{version_spec}"], console_script="mypy", project=project
        ).output_file
    else:
        mypy_pex_bin = None

    task = project.task(name, MypyTask, group="lint")
    task.mypy_pex_bin = mypy_pex_bin
    task.config_file = config_file
    task.additional_args = additional_args
    task.check_tests = check_tests
    task.use_daemon = use_daemon
    task.python_version = python_version
    return task

mypy_stubtest 

mypy_stubtest(
    *,
    name: str = "python.mypy.stubtest",
    project: Project | None = None,
    package: str,
    ignore_missing_stubs: bool = False,
    ignore_positional_only: bool = False,
    allowlist: Path | None = None,
    mypy_config_file: Path | None = None,
    mypy_pex_bin: Path | Property[Path] | None = None
) -> MypyStubtestTask

:param version_spec: If specified, the isort tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/mypy_stubtest_task.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def mypy_stubtest(
    *,
    name: str = "python.mypy.stubtest",
    project: Project | None = None,
    package: str,
    ignore_missing_stubs: bool = False,
    ignore_positional_only: bool = False,
    allowlist: Path | None = None,
    mypy_config_file: Path | None = None,
    mypy_pex_bin: Path | Property[Path] | None = None,
) -> MypyStubtestTask:
    """
    :param version_spec: If specified, the isort tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    project = project or Project.current()
    task = project.task(name, MypyStubtestTask, group="lint")
    task.mypy_pex_bin = mypy_pex_bin
    task.package = package
    task.ignore_missing_stubs = ignore_missing_stubs
    task.ignore_positional_only = ignore_positional_only
    task.allowlist = allowlist
    task.mypy_config_file = mypy_config_file
    return task

publish 

publish(
    *,
    package_index: str,
    distributions: list[Path] | Property[list[Path]],
    skip_existing: bool = False,
    interactive: bool = True,
    name: str = "python.publish",
    group: str | None = "publish",
    project: Project | None = None,
    after: list[Task] | None = None,
    twine_version: str = ">=4.0.2,<5.0.0"
) -> PublishTask

Create a publish task for the specified registry.

Source code in kraken/std/python/tasks/publish_task.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def publish(
    *,
    package_index: str,
    distributions: list[Path] | Property[list[Path]],
    skip_existing: bool = False,
    interactive: bool = True,
    name: str = "python.publish",
    group: str | None = "publish",
    project: Project | None = None,
    after: list[Task] | None = None,
    twine_version: str = ">=4.0.2,<5.0.0",
) -> PublishTask:
    """Create a publish task for the specified registry."""

    project = project or Project.current()
    settings = python_settings(project)
    if package_index not in settings.package_indexes:
        raise ValueError(f"package index {package_index!r} is not defined")

    twine_bin = pex_build(
        "twine", requirements=[f"twine{twine_version}"], console_script="twine", project=project
    ).output_file

    index = settings.package_indexes[package_index]
    task = project.task(name, PublishTask, group=group)
    task.twine_bin = twine_bin
    task.index_upload_url = index.upload_url
    task.index_credentials = index.credentials
    task.distributions = distributions
    task.skip_existing = skip_existing
    task.interactive = interactive
    task.depends_on(*(after or []))
    return task

pycln 

pycln(
    *,
    name: str = "python.pycln",
    project: Project | None = None,
    version_spec: str | None = None
) -> PyclnTasks

Creates two pycln tasks, one to check and another to format. The check task will be grouped under "lint" whereas the format task will be grouped under "fmt".

:param version_spec: If specified, the pycln tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/pycln_task.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def pycln(*, name: str = "python.pycln", project: Project | None = None, version_spec: str | None = None) -> PyclnTasks:
    """Creates two pycln tasks, one to check and another to format. The check task will be grouped under `"lint"`
    whereas the format task will be grouped under `"fmt"`.

    :param version_spec: If specified, the pycln tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    project = project or Project.current()
    if version_spec is not None:
        pycln_bin = pex_build(
            "pycln", requirements=[f"pycln{version_spec}"], console_script="pycln", project=project
        ).output_file.map(str)
    else:
        pycln_bin = Supplier.of("pycln")

    check_task = project.task(f"{name}.check", PyclnTask, group="lint")
    check_task.pycln_bin = pycln_bin
    check_task.check_only = True
    format_task = project.task(name, PyclnTask, group="fmt")
    format_task.pycln_bin = pycln_bin
    return PyclnTasks(check_task, format_task)

pylint 

pylint(
    *,
    name: str = "python.pylint",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_args: (
        Sequence[str] | Property[Sequence[str]]
    ) = (),
    version_spec: str | None = None
) -> PylintTask

:param version_spec: If specified, the pylint tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/pylint_task.py 
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
60
def pylint(
    *,
    name: str = "python.pylint",
    project: Project | None = None,
    config_file: Path | Supplier[Path] | None = None,
    additional_args: Sequence[str] | Property[Sequence[str]] = (),
    version_spec: str | None = None,
) -> PylintTask:
    """
    :param version_spec: If specified, the pylint tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    project = project or Project.current()
    if version_spec is not None:
        pylint_bin = pex_build(
            "pylint", requirements=[f"pylint{version_spec}"], console_script="pylint", project=project
        ).output_file.map(str)
    else:
        pylint_bin = Supplier.of("pylint")

    task = project.task(name, PylintTask, group="lint")
    task.pylint_bin = pylint_bin
    task.config_file = config_file
    task.additional_args = additional_args
    return task

python_settings 

python_settings(
    project: Project | None = None,
    build_system: PythonBuildSystem | None = None,
    source_directory: str | Path | None = None,
    tests_directory: str | Path | None = None,
    lint_enforced_directories: (
        list[str | Path] | None
    ) = None,
    always_use_managed_env: bool | None = None,
    skip_install_if_venv_exists: bool | None = None,
) -> PythonSettings

Read the Python settings for the given or current project and optionally update attributes.

:param project: The project to get the settings for. If not specified, the current project will be used. :environment_handler: If specified, set the :attr:PythonSettings.environment_handler. If a string is specified, the following values are currently supported: "poetry". :param source_directory: The source directory. Defaults to "src". :param tests_directory: The tests directory. Automatically determined if left empty. :param lint_enforced_directories: Any extra directories containing Python files, e.g. bin/, scripts/, and examples/, to be linted alongside the source and tests directories.

Source code in kraken/std/python/settings.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def python_settings(
    project: Project | None = None,
    build_system: PythonBuildSystem | None = None,
    source_directory: str | Path | None = None,
    tests_directory: str | Path | None = None,
    lint_enforced_directories: list[str | Path] | None = None,
    always_use_managed_env: bool | None = None,
    skip_install_if_venv_exists: bool | None = None,
) -> PythonSettings:
    """Read the Python settings for the given or current project and optionally update attributes.

    :param project: The project to get the settings for. If not specified, the current project will be used.
    :environment_handler: If specified, set the :attr:`PythonSettings.environment_handler`. If a string is specified,
        the following values are currently supported: `"poetry"`.
    :param source_directory: The source directory. Defaults to `"src"`.
    :param tests_directory: The tests directory. Automatically determined if left empty.
    :param lint_enforced_directories: Any extra directories containing Python files, e.g. bin/, scripts/, and
        examples/, to be linted alongside the source and tests directories.
    """

    project = project or Project.current()
    settings = project.find_metadata(PythonSettings)
    if settings is None:
        settings = PythonSettings(project)
        project.metadata.append(settings)

    if build_system is None and settings.build_system is None:
        # Autodetect the environment handler.
        build_system = detect_build_system(project.directory)
        if build_system:
            logger.info("Detected Python build system %r for %s", type(build_system).__name__, project)

    if build_system is not None:
        if settings.build_system:
            logger.warning(
                "overwriting existing PythonSettings.environment_handler=%r with %r",
                settings.build_system,
                build_system,
            )
        settings.build_system = build_system

    if source_directory is not None:
        settings.source_directory = Path(source_directory)

    if tests_directory is not None:
        settings.tests_directory = Path(tests_directory)

    if lint_enforced_directories is not None:
        dirs = []
        for directory in lint_enforced_directories:
            directory_path = Path(directory)
            if not directory_path.exists():
                logger.debug(f"skipping specified lint enforced directory {directory_path} as it does not exist")
            elif not directory_path.is_dir():
                logger.warning(f"skipping specified lint enforced directory {directory_path} as it is not a directory")
            else:
                dirs.append(directory_path)
        settings.lint_enforced_directories = dirs

    if always_use_managed_env is not None:
        settings.always_use_managed_env = True

    if skip_install_if_venv_exists is not None:
        settings.skip_install_if_venv_exists = skip_install_if_venv_exists

    return settings

pyupgrade 

pyupgrade(
    *,
    name: str = "python.pyupgrade",
    project: Project | None = None,
    exclude: Collection[Path] = (),
    exclude_patterns: Collection[str] = (),
    keep_runtime_typing: bool = False,
    python_version: str = "3",
    additional_files: Sequence[Path] = (),
    version_spec: str | None = None
) -> PyUpgradeTasks

:param version_spec: If specified, the pyupgrade tool will be installed as a PEX and does not need to be installed into the Python project's virtual env.

Source code in kraken/std/python/tasks/pyupgrade_task.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def pyupgrade(
    *,
    name: str = "python.pyupgrade",
    project: Project | None = None,
    exclude: Collection[Path] = (),
    exclude_patterns: Collection[str] = (),
    keep_runtime_typing: bool = False,
    python_version: str = "3",
    additional_files: Sequence[Path] = (),
    version_spec: str | None = None,
) -> PyUpgradeTasks:
    """
    :param version_spec: If specified, the pyupgrade tool will be installed as a PEX and does not need to be installed
        into the Python project's virtual env.
    """

    project = project or Project.current()

    if version_spec is not None:
        pyupgrade_bin = pex_build(
            "pyupgrade", requirements=[f"pyupgrade{version_spec}"], console_script="pyupgrade", project=project
        ).output_file.map(str)
    else:
        pyupgrade_bin = Supplier.of("pyupgrade")

    settings = python_settings(project)

    directories = list(additional_files)
    directories.append(project.directory / settings.source_directory)
    test_directory = settings.get_tests_directory()
    if test_directory is not None:
        directories.append(project.directory / test_directory)
    files = {f.resolve() for p in directories for f in Path(p).glob("**/*.py")}
    exclude = [e.resolve() for e in exclude]
    filtered_files = [
        f
        for f in files
        if not any(f.is_relative_to(i) for i in exclude) and not any(f.match(p) for p in exclude_patterns)
    ]

    check_task = project.task(f"{name}.check", PyUpgradeCheckTask, group="lint")
    check_task.pyupgrade_bin = pyupgrade_bin
    check_task.additional_files = filtered_files
    check_task.keep_runtime_typing = keep_runtime_typing
    check_task.python_version = python_version

    format_task = project.task(name, PyUpgradeTask, group="fmt")
    format_task.pyupgrade_bin = pyupgrade_bin
    format_task.additional_files = filtered_files
    format_task.keep_runtime_typing = keep_runtime_typing
    format_task.python_version = python_version

    return PyUpgradeTasks(check_task, format_task)

update_lockfile_task 

update_lockfile_task(
    *,
    name: str = "python.update",
    group: str | None = "update",
    project: Project | None = None
) -> UpdateLockfileTask

Creates an update task for the given project.

The update task relies on the build system configured in the Python project settings.

Source code in kraken/std/python/tasks/update_lockfile_task.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def update_lockfile_task(
    *,
    name: str = "python.update",
    group: str | None = "update",
    project: Project | None = None,
) -> UpdateLockfileTask:
    """Creates an update task for the given project.

    The update task relies on the build system configured in the Python project settings."""

    project = project or Project.current()
    task = project.task(name, UpdateLockfileTask, group=group)
    task.settings = python_settings(project)
    task.build_system = Supplier.of_callable(lambda: python_settings(project).build_system)
    task.pyproject_toml = project.directory / "pyproject.toml"
    return task