17573005466

Committed 09 Sep 2025 05:43AM UTC coverage: 91.392% (-0.1%) from 91.489%

Build # 17573005466

Build Type

push

github

Committed by

waketzheng

Commit Message

chore: upgrade deps

Run Details

860 of 941 relevant lines covered (91.39%)

5.47 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.38

/fast_dev_cli/cli.py

from __future__ import annotations

import contextlib
import functools
import importlib.metadata as importlib_metadata
import os
import re
import shlex
import shutil
import subprocess  # nosec:B404
import sys
from functools import cached_property
from pathlib import Path
from typing import (
    TYPE_CHECKING,
    Any,
    Literal,
    # Optional is required by Option generated by typer
    Optional,
    cast,
    get_args,
    overload,
)

import typer
from click import UsageError
from typer import Exit, Option, echo, secho
from typer.models import ArgumentInfo, OptionInfo

try:
    from . import __version__
except ImportError:  # pragma: no cover
    from importlib import import_module as _import  # For local unittest

    __version__ = _import(Path(__file__).parent.name).__version__

if sys.version_info >= (3, 11):  # pragma: no cover
    from enum import StrEnum

    import tomllib
else:  # pragma: no cover
    from enum import Enum

    import tomli as tomllib

    class StrEnum(str, Enum):
        __str__ = str.__str__


if TYPE_CHECKING:
    if sys.version_info >= (3, 11):
        from typing import Self
    else:
        from typing_extensions import Self

cli = typer.Typer(no_args_is_help=True)
DryOption = Option(False, "--dry", help="Only print, not really run shell command")
TOML_FILE = "pyproject.toml"
ToolName = Literal["poetry", "pdm", "uv"]
ToolOption = Option(
    "auto", "--tool", help="Explicit declare manage tool (default to auto detect)"
)


class FastDevCliError(Exception):
    """Basic exception of this library, all custom exceptions inherit from it"""


class ShellCommandError(FastDevCliError):
    """Raise if cmd command returncode is not zero"""


class ParseError(FastDevCliError):
    """Raise this if parse dependence line error"""


class EnvError(FastDevCliError):
    """Raise when expected to be managed by poetry, but toml file not found."""


def poetry_module_name(name: str) -> str:
    """Get module name that generated by `poetry new`"""
    try:
        from packaging.utils import canonicalize_name
    except ImportError:

        def canonicalize_name(s: str) -> str:  # type:ignore[misc]
            return re.sub(r"[-_.]+", "-", s)

    return canonicalize_name(name).replace("-", "_").replace(" ", "_")


def is_emoji(char: str) -> bool:
    try:
        import emoji
    except ImportError:
        pass
    else:
        return emoji.is_emoji(char)
    if re.match(r"[\w\d\s]", char):
        return False
    return not "\u4e00" <= char <= "\u9fff"  # Chinese character


def yellow_warn(msg: str) -> None:
    secho(msg, fg="yellow")


def load_bool(name: str, default: bool = False) -> bool:
    if not (v := os.getenv(name)):
        return default
    if (lower := v.lower()) in ("0", "false", "f", "off", "no", "n"):
        return False
    elif lower in ("1", "true", "t", "on", "yes", "y"):
        return True
    secho(f"WARNING: can not convert value({v!r}) of {name} to bool!")
    return default


def is_venv() -> bool:
    """Whether in a virtual environment(also work for poetry)"""
    return hasattr(sys, "real_prefix") or (
        hasattr(sys, "base_prefix") and sys.base_prefix != sys.prefix
    )


def _run_shell(cmd: list[str] | str, **kw: Any) -> subprocess.CompletedProcess[str]:
    if isinstance(cmd, str):
        kw.setdefault("shell", True)
    return subprocess.run(cmd, **kw)  # nosec:B603


def run_and_echo(
    cmd: str, *, dry: bool = False, verbose: bool = True, **kw: Any
) -> int:
    """Run shell command with subprocess and print it"""
    if verbose:
        echo(f"--> {cmd}")
    if dry:
        return 0
    command: list[str] | str = cmd
    if "shell" not in kw and not (set(cmd) & {"|", ">", "&"}):
        command = shlex.split(cmd)
    return _run_shell(command, **kw).returncode


def check_call(cmd: str) -> bool:
    r = _run_shell(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return r.returncode == 0


def capture_cmd_output(
    command: list[str] | str, *, raises: bool = False, **kw: Any
) -> str:
    if isinstance(command, str) and not kw.get("shell"):
        command = shlex.split(command)
    r = _run_shell(command, capture_output=True, encoding="utf-8", **kw)
    if raises and r.returncode != 0:
        raise ShellCommandError(r.stderr)
    return r.stdout.strip() or r.stderr


def _parse_version(line: str, pattern: re.Pattern[str]) -> str:
    return pattern.sub("", line).split("#")[0].strip(" '\"")


def read_version_from_file(
    package_name: str, work_dir: Path | None = None, toml_text: str | None = None
) -> str:
    if not package_name and toml_text:
        pattern = re.compile(r"version\s*=")
        for line in toml_text.splitlines():
            if pattern.match(line):
                return _parse_version(line, pattern)
    version_file = BumpUp.parse_filename(toml_text, work_dir, package_name)
    if version_file == TOML_FILE:
        if toml_text is None:
            toml_text = Project.load_toml_text()
        context = tomllib.loads(toml_text)
        with contextlib.suppress(KeyError):
            return cast(str, context["project"]["version"])
        with contextlib.suppress(KeyError):  # Poetry V1
            return cast(str, context["tool"]["poetry"]["version"])
        secho(f"WARNING: can not find 'version' item in {version_file}!")
        return "0.0.0"
    pattern = re.compile(r"__version__\s*=")
    for line in Path(version_file).read_text("utf-8").splitlines():
        if pattern.match(line):
            return _parse_version(line, pattern)
    # TODO: remove or refactor the following lines.
    if work_dir is None:
        work_dir = Project.get_work_dir()
    package_dir = work_dir / package_name
    if (
        not (init_file := package_dir / "__init__.py").exists()
        and not (init_file := work_dir / "src" / package_name / init_file.name).exists()
        and not (init_file := work_dir / "app" / init_file.name).exists()
    ):
        secho("WARNING: __init__.py file does not exist!")
        return "0.0.0"

    pattern = re.compile(r"__version__\s*=")
    for line in init_file.read_text("utf-8").splitlines():
        if pattern.match(line):
            return _parse_version(line, pattern)
    secho(f"WARNING: can not find '__version__' var in {init_file}!")
    return "0.0.0"


@overload
def get_current_version(
    verbose: bool = False,
    is_poetry: bool | None = None,
    package_name: str | None = None,
    *,
    check_version: Literal[False] = False,
) -> str: ...


@overload
def get_current_version(
    verbose: bool = False,
    is_poetry: bool | None = None,
    package_name: str | None = None,
    *,
    check_version: Literal[True] = True,
) -> tuple[bool, str]: ...


def get_current_version(
    verbose: bool = False,
    is_poetry: bool | None = None,
    package_name: str | None = None,
    *,
    check_version: bool = False,
) -> str | tuple[bool, str]:
    if is_poetry is True or Project.manage_by_poetry():
        cmd = ["poetry", "version", "-s"]
        if verbose:
            echo(f"--> {' '.join(cmd)}")
        if out := capture_cmd_output(cmd, raises=True):
            out = out.splitlines()[-1].strip().split()[-1]
        if check_version:
            return True, out
        return out
    toml_text = work_dir = None
    if package_name is None:
        work_dir = Project.get_work_dir()
        toml_text = Project.load_toml_text()
        doc = tomllib.loads(toml_text)
        project_name = doc.get("project", {}).get("name", work_dir.name)
        package_name = re.sub(r"[- ]", "_", project_name)
    local_version = read_version_from_file(package_name, work_dir, toml_text)
    try:
        installed_version = importlib_metadata.version(package_name)
    except importlib_metadata.PackageNotFoundError:
        installed_version = ""
    current_version = local_version or installed_version
    if not current_version:
        raise FastDevCliError(f"Failed to get current version of {package_name!r}")
    if check_version:
        is_conflict = bool(local_version) and local_version != installed_version
        return is_conflict, current_version
    return current_version


def _ensure_bool(value: bool | OptionInfo) -> bool:
    if not isinstance(value, bool):
        value = getattr(value, "default", False)
    return value


def _ensure_str(value: str | OptionInfo | None) -> str:
    if not isinstance(value, str):
        value = getattr(value, "default", "")
    return value


def exit_if_run_failed(
    cmd: str,
    env: dict[str, str] | None = None,
    _exit: bool = False,
    dry: bool = False,
    **kw: Any,
) -> subprocess.CompletedProcess[str]:
    run_and_echo(cmd, dry=True)
    if _ensure_bool(dry):
        return subprocess.CompletedProcess("", 0)
    if env is not None:
        env = {**os.environ, **env}
    r = _run_shell(cmd, env=env, **kw)
    if rc := r.returncode:
        if _exit:
            sys.exit(rc)
        raise Exit(rc)
    return r


class DryRun:
    def __init__(self, _exit: bool = False, dry: bool = False) -> None:
        self.dry = _ensure_bool(dry)
        self._exit = _exit

    def gen(self) -> str:
        raise NotImplementedError

    def run(self) -> None:
        exit_if_run_failed(self.gen(), _exit=self._exit, dry=self.dry)


class BumpUp(DryRun):
    class PartChoices(StrEnum):
        patch = "patch"
        minor = "minor"
        major = "major"

    def __init__(
        self,
        commit: bool,
        part: str,
        filename: str | None = None,
        dry: bool = False,
        no_sync: bool = False,
        emoji: bool | None = None,
    ) -> None:
        self.commit = commit
        self.part = part
        if filename is None:
            filename = self.parse_filename()
        self.filename = filename
        self._no_sync = no_sync
        self._emoji = emoji
        super().__init__(dry=dry)

    @staticmethod
    def get_last_commit_message(raises: bool = False) -> str:
        cmd = 'git show --pretty=format:"%s" -s HEAD'
        return capture_cmd_output(cmd, raises=raises)

    @classmethod
    def should_add_emoji(cls) -> bool:
        """
        If last commit message is startswith emoji,
        add a ⬆️ flag at the prefix of bump up commit message.
        """
        try:
            first_char = cls.get_last_commit_message(raises=True)[0]
        except (IndexError, ShellCommandError):
            return False
        else:
            return is_emoji(first_char)

    @staticmethod
    def parse_filename(
        toml_text: str | None = None,
        work_dir: Path | None = None,
        package_name: str | None = None,
    ) -> str:
        if toml_text is None:
            toml_text = Project.load_toml_text()
        context = tomllib.loads(toml_text)
        by_version_plugin = False
        try:
            ver = context["project"]["version"]
        except KeyError:
            pass
        else:
            if isinstance(ver, str):
                if ver in ("0", "0.0.0"):
                    by_version_plugin = True
                elif re.match(r"\d+\.\d+\.\d+", ver):
                    return TOML_FILE
        if not by_version_plugin:
            try:
                version_value = context["tool"]["poetry"]["version"]
            except KeyError:
                if not Project.manage_by_poetry():
                    if work_dir is None:
                        work_dir = Project.get_work_dir()
                    for tool in ("pdm", "hatch"):
                        with contextlib.suppress(KeyError):
                            version_path = cast(
                                str, context["tool"][tool]["version"]["path"]
                            )
                            if (
                                Path(version_path).exists()
                                or work_dir.joinpath(version_path).exists()
                            ):
                                return version_path
                    # version = { source = "file", path = "fast_dev_cli/__init__.py" }
                    v_key = "version = "
                    p_key = 'path = "'
                    for line in toml_text.splitlines():
                        if not line.startswith(v_key):
                            continue
                        if p_key in (value := line.split(v_key, 1)[-1].split("#")[0]):
                            filename = value.split(p_key, 1)[-1].split('"')[0]
                            if work_dir.joinpath(filename).exists():
                                return filename
            else:
                by_version_plugin = version_value in ("0", "0.0.0", "init")
        if by_version_plugin:
            try:
                package_item = context["tool"]["poetry"]["packages"]
            except KeyError:
                try:
                    project_name = context["project"]["name"]
                except KeyError:
                    packages = []
                else:
                    packages = [(poetry_module_name(project_name), "")]
            else:
                packages = [
                    (j, i.get("from", ""))
                    for i in package_item
                    if (j := i.get("include"))
                ]
            # In case of managed by `poetry-plugin-version`
            cwd = Path.cwd()
            pattern = re.compile(r"__version__\s*=\s*['\"]")
            ds: list[Path] = []
            if package_name is not None:
                packages.insert(0, (package_name, ""))
            for package_name, source_dir in packages:
                ds.append(cwd / package_name)
                ds.append(cwd / "src" / package_name)
                if source_dir and source_dir != "src":
                    ds.append(cwd / source_dir / package_name)
            module_name = poetry_module_name(cwd.name)
            ds.extend([cwd / module_name, cwd / "src" / module_name, cwd])
            for d in ds:
                init_file = d / "__init__.py"
                if (
                    init_file.exists() and pattern.search(init_file.read_text("utf8"))
                ) or (
                    (init_file := init_file.with_name("__version__.py")).exists()
                    and pattern.search(init_file.read_text("utf8"))
                ):
                    break
            else:
                raise ParseError("Version file not found! Where are you now?")
            return os.path.relpath(init_file, cwd)

        return TOML_FILE

    def get_part(self, s: str) -> str:
        choices: dict[str, str] = {}
        for i, p in enumerate(self.PartChoices, 1):
            v = str(p)
            choices.update({str(i): v, v: v})
        try:
            return choices[s]
        except KeyError as e:
            echo(f"Invalid part: {s!r}")
            raise Exit(1) from e

    def gen(self) -> str:
        should_sync, _version = get_current_version(check_version=True)
        filename = self.filename
        echo(f"Current version(@{filename}): {_version}")
        if self.part:
            part = self.get_part(self.part)
        else:
            part = "patch"
            if a := input("Which one?").strip():
                part = self.get_part(a)
        self.part = part
        parse = r'--parse "(?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)"'
        cmd = f'bumpversion {parse} --current-version="{_version}" {part} {filename}'
        if self.commit:
            if part != "patch":
                cmd += " --tag"
            cmd += " --commit"
            if self._emoji or (self._emoji is None and self.should_add_emoji()):
                cmd += " --message-emoji=1"
            if not load_bool("DONT_GIT_PUSH"):
                cmd += " && git push && git push --tags && git log -1"
        else:
            cmd += " --allow-dirty"
        if should_sync and not self._no_sync and (sync := Project.get_sync_command()):
            cmd = f"{sync} && " + cmd
        return cmd

    def run(self) -> None:
        super().run()
        if not self.commit and not self.dry:
            new_version = get_current_version(True)
            echo(new_version)
            if self.part != "patch":
                echo("You may want to pin tag by `fast tag`")


@cli.command()
def version() -> None:
    """Show the version of this tool"""
    echo("Fast Dev Cli Version: " + typer.style(__version__, fg=typer.colors.BLUE))
    with contextlib.suppress(FileNotFoundError, KeyError):
        toml_text = Project.load_toml_text()
        doc = tomllib.loads(toml_text)
        version_file = doc["tool"]["pdm"]["version"]["path"]
        text = Project.get_work_dir().joinpath(version_file).read_text()
        varname = "__version__"
        for line in text.splitlines():
            if line.strip().startswith(varname):
                value = line.split("=", 1)[-1].strip().strip('"').strip("'")
                styled = typer.style(value, bold=True)
                echo(f"Version value in {version_file}: " + styled)
                break


@cli.command(name="bump")
def bump_version(
    part: BumpUp.PartChoices,
    commit: bool = Option(
        False, "--commit", "-c", help="Whether run `git commit` after version changed"
    ),
    emoji: Optional[bool] = Option(
        None, "--emoji", help="Whether add emoji prefix to commit message"
    ),
    no_sync: bool = Option(
        False, "--no-sync", help="Do not run sync command to update version"
    ),
    dry: bool = DryOption,
) -> None:
    """Bump up version string in pyproject.toml"""
    if emoji is not None:
        emoji = _ensure_bool(emoji)
    return BumpUp(
        _ensure_bool(commit),
        getattr(part, "value", part),
        no_sync=_ensure_bool(no_sync),
        emoji=emoji,
        dry=dry,
    ).run()


def bump() -> None:
    part, commit = "", False
    if args := sys.argv[2:]:
        if "-c" in args or "--commit" in args:
            commit = True
        for a in args:
            if not a.startswith("-"):
                part = a
                break
    return BumpUp(commit, part, no_sync="--no-sync" in args, dry="--dry" in args).run()


class Project:
    path_depth = 5
    _tool: ToolName | None = None

    @staticmethod
    def is_poetry_v2(text: str) -> bool:
        return 'build-backend = "poetry' in text

    @staticmethod
    def work_dir(
        name: str, parent: Path, depth: int, be_file: bool = False
    ) -> Path | None:
        for _ in range(depth):
            if (f := parent.joinpath(name)).exists():
                if be_file:
                    return f
                return parent
            parent = parent.parent
        return None

    @classmethod
    def get_work_dir(
        cls: type[Self],
        name: str = TOML_FILE,
        cwd: Path | None = None,
        allow_cwd: bool = False,
        be_file: bool = False,
    ) -> Path:
        cwd = cwd or Path.cwd()
        if d := cls.work_dir(name, cwd, cls.path_depth, be_file):
            return d
        if allow_cwd:
            return cls.get_root_dir(cwd)
        raise EnvError(f"{name} not found! Make sure this is a python project.")

    @classmethod
    def load_toml_text(cls: type[Self], name: str = TOML_FILE) -> str:
        toml_file = cls.get_work_dir(name, be_file=True)
        return toml_file.read_text("utf8")

    @classmethod
    def manage_by_poetry(cls: type[Self], cache: bool = False) -> bool:
        return cls.get_manage_tool(cache=cache) == "poetry"

    @classmethod
    def get_manage_tool(cls: type[Self], cache: bool = False) -> ToolName | None:
        if cache and cls._tool:
            return cls._tool
        try:
            text = cls.load_toml_text()
        except EnvError:
            return None
        with contextlib.suppress(KeyError, tomllib.TOMLDecodeError):
            doc = tomllib.loads(text)
            backend = doc["build-system"]["build-backend"]
            if "poetry" in backend:
                cls._tool = "poetry"
                return cls._tool
            work_dir = cls.get_work_dir(allow_cwd=True)
            uv_lock_exists = Path(work_dir, "uv.lock").exists()
            if "pdm" in backend:
                cls._tool = "pdm"
                if not Path(work_dir, "pdm.lock").exists() and (
                    uv_lock_exists or "[tool.uv]" in text
                ):
                    cls._tool = "uv"
                return cls._tool
            elif uv_lock_exists:
                cls._tool = "uv"
                return cls._tool
        for name in get_args(ToolName):
            if f"[tool.{name}]" in text:
                cls._tool = cast(ToolName, name)
                return cls._tool
        # Poetry 2.0 default to not include the '[tool.poetry]' section
        if cls.is_poetry_v2(text):
            cls._tool = "poetry"
            return cls._tool
        return None

    @staticmethod
    def python_exec_dir() -> Path:
        return Path(sys.executable).parent

    @classmethod
    def get_root_dir(cls: type[Self], cwd: Path | None = None) -> Path:
        root = cwd or Path.cwd()
        venv_parent = cls.python_exec_dir().parent.parent
        if root.is_relative_to(venv_parent):
            root = venv_parent
        return root

    @classmethod
    def is_pdm_project(cls, strict: bool = True, cache: bool = False) -> bool:
        if cls.get_manage_tool(cache=cache) != "pdm":
            return False
        if strict:
            lock_file = cls.get_work_dir() / "pdm.lock"
            return lock_file.exists()
        return True

    @classmethod
    def get_sync_command(cls, prod: bool = True, doc: dict | None = None) -> str:
        if cls.is_pdm_project():
            return "pdm sync" + " --prod" * prod
        elif cls.manage_by_poetry(cache=True):
            cmd = "poetry install"
            if prod:
                if doc is None:
                    doc = tomllib.loads(cls.load_toml_text())
                if doc.get("project", {}).get("dependencies") or any(
                    i != "python"
                    for i in doc.get("tool", {})
                    .get("poetry", {})
                    .get("dependencies", [])
                ):
                    cmd += " --only=main"
            return cmd
        elif cls.get_manage_tool(cache=True) == "uv":
            return "uv sync --inexact" + " --no-dev" * prod
        return ""

    @classmethod
    def sync_dependencies(cls, prod: bool = True) -> None:
        if cmd := cls.get_sync_command():
            run_and_echo(cmd)


class UpgradeDependencies(Project, DryRun):
    def __init__(
        self, _exit: bool = False, dry: bool = False, tool: ToolName = "poetry"
    ) -> None:
        super().__init__(_exit, dry)
        self._tool = tool

    class DevFlag(StrEnum):
        new = "[tool.poetry.group.dev.dependencies]"
        old = "[tool.poetry.dev-dependencies]"

    @staticmethod
    def parse_value(version_info: str, key: str) -> str:
        """Pick out the value for key in version info.

        Example::
            >>> s= 'typer = {extras = ["all"], version = "^0.9.0", optional = true}'
            >>> UpgradeDependencies.parse_value(s, 'extras')
            'all'
            >>> UpgradeDependencies.parse_value(s, 'optional')
            'true'
            >>> UpgradeDependencies.parse_value(s, 'version')
            '^0.9.0'
        """
        sep = key + " = "
        rest = version_info.split(sep, 1)[-1].strip(" =")
        if rest.startswith("["):
            rest = rest[1:].split("]")[0]
        elif rest.startswith('"'):
            rest = rest[1:].split('"')[0]
        else:
            rest = rest.split(",")[0].split("}")[0]
        return rest.strip().replace('"', "")

    @staticmethod
    def no_need_upgrade(version_info: str, line: str) -> bool:
        if (v := version_info.replace(" ", "")).startswith("{url="):
            echo(f"No need to upgrade for: {line}")
            return True
        if (f := "version=") in v:
            v = v.split(f)[1].strip('"').split('"')[0]
        if v == "*":
            echo(f"Skip wildcard line: {line}")
            return True
        elif v == "[":
            echo(f"Skip complex dependence: {line}")
            return True
        elif v.startswith(">") or v.startswith("<") or v[0].isdigit():
            echo(f"Ignore bigger/smaller/equal: {line}")
            return True
        return False

    @classmethod
    def build_args(
        cls: type[Self], package_lines: list[str]
    ) -> tuple[list[str], dict[str, list[str]]]:
        args: list[str] = []  # ['typer[all]', 'fastapi']
        specials: dict[str, list[str]] = {}  # {'--platform linux': ['gunicorn']}
        for no, line in enumerate(package_lines, 1):
            if (
                not (m := line.strip())
                or m.startswith("#")
                or m == "]"
                or (m.startswith("{") and m.strip(",").endswith("}"))
            ):
                continue
            try:
                package, version_info = m.split("=", 1)
            except ValueError as e:
                raise ParseError(f"Failed to separate by '='@line {no}: {m}") from e
            if (package := package.strip()).lower() == "python":
                continue
            if cls.no_need_upgrade(version_info := version_info.strip(' "'), line):
                continue
            if (extras_tip := "extras") in version_info:
                package += "[" + cls.parse_value(version_info, extras_tip) + "]"
            item = f'"{package}@latest"'
            key = None
            if (pf := "platform") in version_info:
                platform = cls.parse_value(version_info, pf)
                key = f"--{pf}={platform}"
            if (sc := "source") in version_info:
                source = cls.parse_value(version_info, sc)
                key = ("" if key is None else (key + " ")) + f"--{sc}={source}"
            if "optional = true" in version_info:
                key = ("" if key is None else (key + " ")) + "--optional"
            if key is not None:
                specials[key] = specials.get(key, []) + [item]
            else:
                args.append(item)
        return args, specials

    @classmethod
    def should_with_dev(cls: type[Self]) -> bool:
        text = cls.load_toml_text()
        return cls.DevFlag.new in text or cls.DevFlag.old in text

    @staticmethod
    def parse_item(toml_str: str) -> list[str]:
        lines: list[str] = []
        for line in toml_str.splitlines():
            if (line := line.strip()).startswith("["):
                if lines:
                    break
            elif line:
                lines.append(line)
        return lines

    @classmethod
    def get_args(
        cls: type[Self], toml_text: str | None = None
    ) -> tuple[list[str], list[str], list[list[str]], str]:
        if toml_text is None:
            toml_text = cls.load_toml_text()
        main_title = "[tool.poetry.dependencies]"
        if (no_main_deps := main_title not in toml_text) and not cls.is_poetry_v2(
            toml_text
        ):
            raise EnvError(
                f"{main_title} not found! Make sure this is a poetry project."
            )
        text = toml_text.split(main_title)[-1]
        dev_flag = "--group dev"
        new_flag, old_flag = cls.DevFlag.new, cls.DevFlag.old
        if (dev_title := getattr(new_flag, "value", new_flag)) not in text:
            dev_title = getattr(old_flag, "value", old_flag)  # For poetry<=1.2
            dev_flag = "--dev"
        others: list[list[str]] = []
        try:
            main_toml, dev_toml = text.split(dev_title)
        except ValueError:
            dev_toml = ""
            main_toml = text
        mains = [] if no_main_deps else cls.parse_item(main_toml)
        devs = cls.parse_item(dev_toml)
        prod_packs, specials = cls.build_args(mains)
        if specials:
            others.extend([[k] + v for k, v in specials.items()])
        dev_packs, specials = cls.build_args(devs)
        if specials:
            others.extend([[k] + v + [dev_flag] for k, v in specials.items()])
        return prod_packs, dev_packs, others, dev_flag

    @classmethod
    def gen_cmd(cls: type[Self]) -> str:
        main_args, dev_args, others, dev_flags = cls.get_args()
        return cls.to_cmd(main_args, dev_args, others, dev_flags)

    @staticmethod
    def to_cmd(
        main_args: list[str],
        dev_args: list[str],
        others: list[list[str]],
        dev_flags: str,
    ) -> str:
        command = "poetry add "
        _upgrade = ""
        if main_args:
            _upgrade = command + " ".join(main_args)
        if dev_args:
            if _upgrade:
                _upgrade += " && "
            _upgrade += command + dev_flags + " " + " ".join(dev_args)
        for single in others:
            _upgrade += f" && poetry add {' '.join(single)}"
        return _upgrade

    def gen(self) -> str:
        if self._tool == "uv":
            up = "uv lock --upgrade --verbose"
            deps = "uv sync --inexact --frozen --all-groups --all-extras"
            return f"{up} && {deps}"
        elif self._tool == "pdm":
            return "pdm update --verbose && pdm sync -G :all --frozen"
        return self.gen_cmd() + " && poetry lock && poetry update"


@cli.command()
def upgrade(
    tool: str = ToolOption,
    dry: bool = DryOption,
) -> None:
    """Upgrade dependencies in pyproject.toml to latest versions"""
    if not (tool := _ensure_str(tool)) or tool == ToolOption.default:
        tool = Project.get_manage_tool() or "uv"
    if tool in get_args(ToolName):
        UpgradeDependencies(dry=dry, tool=cast(ToolName, tool)).run()
    else:
        secho(f"Unknown tool {tool!r}", fg=typer.colors.YELLOW)
        raise typer.Exit(1)


class GitTag(DryRun):
    def __init__(self, message: str, dry: bool, no_sync: bool = False) -> None:
        self.message = message
        self._no_sync = no_sync
        super().__init__(dry=dry)

    @staticmethod
    def has_v_prefix() -> bool:
        return "v" in capture_cmd_output("git tag")

    def should_push(self) -> bool:
        return "git push" in self.git_status

    def gen(self) -> str:
        should_sync, _version = get_current_version(verbose=False, check_version=True)
        if self.has_v_prefix():
            # Add `v` at prefix to compare with bumpversion tool
            _version = "v" + _version
        cmd = f"git tag -a {_version} -m {self.message!r} && git push --tags"
        if self.should_push():
            cmd += " && git push"
        if should_sync and not self._no_sync and (sync := Project.get_sync_command()):
            cmd = f"{sync} && " + cmd
        return cmd

    @cached_property
    def git_status(self) -> str:
        return capture_cmd_output("git status")

    def mark_tag(self) -> bool:
        if not re.search(r"working (tree|directory) clean", self.git_status) and (
            "无文件要提交，干净的工作区" not in self.git_status
        ):
            run_and_echo("git status")
            echo("ERROR: Please run git commit to make sure working tree is clean!")
            return False
        return bool(super().run())

    def run(self) -> None:
        if self.mark_tag() and not self.dry:
            echo("You may want to publish package:\n poetry publish --build")


@cli.command()
def tag(
    message: str = Option("", "-m", "--message"),
    no_sync: bool = Option(
        False, "--no-sync", help="Do not run sync command to update version"
    ),
    dry: bool = DryOption,
) -> None:
    """Run shell command: git tag -a <current-version-in-pyproject.toml> -m {message}"""
    GitTag(message, dry=dry, no_sync=_ensure_bool(no_sync)).run()


class LintCode(DryRun):
    def __init__(
        self,
        args: list[str] | str | None,
        check_only: bool = False,
        _exit: bool = False,
        dry: bool = False,
        bandit: bool = False,
        skip_mypy: bool = False,
        dmypy: bool = False,
        tool: str = ToolOption.default,
        prefix: bool = False,
    ) -> None:
        self.args = args
        self.check_only = check_only
        self._bandit = bandit
        self._skip_mypy = skip_mypy
        self._use_dmypy = dmypy
        self._tool = tool
        self._prefix = prefix
        super().__init__(_exit, dry)

    @staticmethod
    def check_lint_tool_installed() -> bool:
        return check_call("ruff --version")

    @staticmethod
    def missing_mypy_exec() -> bool:
        return shutil.which("mypy") is None

    @staticmethod
    def prefer_dmypy(paths: str, tools: list[str], use_dmypy: bool = False) -> bool:
        return (
            paths == "."
            and any(t.startswith("mypy") for t in tools)
            and (use_dmypy or load_bool("FASTDEVCLI_DMYPY"))
        )

    @staticmethod
    def get_package_name() -> str:
        root = Project.get_work_dir(allow_cwd=True)
        module_name = root.name.replace("-", "_").replace(" ", "_")
        package_maybe = (module_name, "src")
        for name in package_maybe:
            if root.joinpath(name).is_dir():
                return name
        return "."

    @classmethod
    def to_cmd(
        cls: type[Self],
        paths: str = ".",
        check_only: bool = False,
        bandit: bool = False,
        skip_mypy: bool = False,
        use_dmypy: bool = False,
        tool: str = ToolOption.default,
        with_prefix: bool = False,
    ) -> str:
        if paths != "." and all(i.endswith(".html") for i in paths.split()):
            return f"prettier -w {paths}"
        cmd = ""
        tools = ["ruff format", "ruff check --extend-select=I,B,SIM --fix", "mypy"]
        if check_only:
            tools[0] += " --check"
        if check_only or load_bool("NO_FIX"):
            tools[1] = tools[1].replace(" --fix", "")
        if skip_mypy or load_bool("SKIP_MYPY") or load_bool("FASTDEVCLI_NO_MYPY"):
            # Sometimes mypy is too slow
            tools = tools[:-1]
        elif load_bool("IGNORE_MISSING_IMPORTS"):
            tools[-1] += " --ignore-missing-imports"
        lint_them = " && ".join(
            "{0}{" + str(i) + "} {1}" for i in range(2, len(tools) + 2)
        )
        if ruff_exists := cls.check_lint_tool_installed():
            # `ruff <command>` get the same result with `pdm run ruff <command>`
            # While `mypy .`(installed global and env not activated),
            #   does not the same as `pdm run mypy .`
            lint_them = " && ".join(
                ("" if tool.startswith("ruff") else "{0}")
                + (
                    "{%d} {1}" % i  # noqa: UP031
                )
                for i, tool in enumerate(tools, 2)
            )
        prefix = ""
        should_run_by_tool = with_prefix
        if not should_run_by_tool:
            if is_venv() and Path(sys.argv[0]).parent != Path.home().joinpath(
                ".local/bin"
            ):
                if not ruff_exists:
                    should_run_by_tool = True
                    command = "pipx install ruff"
                    if shutil.which("pipx") is None:
                        ensure_pipx = "pip install --user pipx\n  pipx ensurepath\n  "
                        command = ensure_pipx + command
                    yellow_warn(
                        "You may need to run the following command"
                        f" to install ruff:\n\n  {command}\n"
                    )
                elif "mypy" in str(tools) and cls.missing_mypy_exec():
                    should_run_by_tool = True
                    if check_call('python -c "import fast_dev_cli"'):
                        command = 'python -m pip install -U "fast-dev-cli"'
                        yellow_warn(
                            "You may need to run the following command"
                            f" to install lint tools:\n\n  {command}\n"
                        )
            else:
                should_run_by_tool = True
        if should_run_by_tool and tool:
            if tool == ToolOption.default:
                tool = Project.get_manage_tool() or ""
            if tool:
                prefix = (
                    bin_dir
                    if tool == "uv" and Path(bin_dir := ".venv/bin/").exists()
                    else (tool + " run ")
                )
        if cls.prefer_dmypy(paths, tools, use_dmypy=use_dmypy):
            tools[-1] = "dmypy run"
        cmd += lint_them.format(prefix, paths, *tools)
        if bandit or load_bool("FASTDEVCLI_BANDIT"):
            command = prefix + "bandit"
            if Path("pyproject.toml").exists():
                toml_text = Project.load_toml_text()
                if "[tool.bandit" in toml_text:
                    command += " -c pyproject.toml"
            if paths == "." and " -c " not in command:
                paths = cls.get_package_name()
            command += f" -r {paths}"
            cmd += " && " + command
        return cmd

    def gen(self) -> str:
        if isinstance(args := self.args, str):
            args = args.split()
        paths = " ".join(map(str, args)) if args else "."
        return self.to_cmd(
            paths,
            self.check_only,
            self._bandit,
            self._skip_mypy,
            self._use_dmypy,
            tool=self._tool,
            with_prefix=self._prefix,
        )


def parse_files(args: list[str] | tuple[str, ...]) -> list[str]:
    return [i for i in args if not i.startswith("-")]


def lint(
    files: list[str] | str | None = None,
    dry: bool = False,
    bandit: bool = False,
    skip_mypy: bool = False,
    dmypy: bool = False,
    tool: str = ToolOption.default,
    prefix: bool = False,
) -> None:
    if files is None:
        files = parse_files(sys.argv[1:])
    if files and files[0] == "lint":
        files = files[1:]
    LintCode(
        files,
        dry=dry,
        skip_mypy=skip_mypy,
        bandit=bandit,
        dmypy=dmypy,
        tool=tool,
        prefix=prefix,
    ).run()


def check(
    files: list[str] | str | None = None,
    dry: bool = False,
    bandit: bool = False,
    skip_mypy: bool = False,
    dmypy: bool = False,
    tool: str = ToolOption.default,
) -> None:
    LintCode(
        files,
        check_only=True,
        _exit=True,
        dry=dry,
        bandit=bandit,
        skip_mypy=skip_mypy,
        dmypy=dmypy,
        tool=tool,
    ).run()


@cli.command(name="lint")
def make_style(
    files: Optional[list[str]] = typer.Argument(default=None),  # noqa:B008
    check_only: bool = Option(False, "--check-only", "-c"),
    bandit: bool = Option(False, "--bandit", help="Run `bandit -r <package_dir>`"),
    prefix: bool = Option(
        False,
        "--prefix",
        help="Run lint command with tool prefix, e.g.: pdm run ruff ...",
    ),
    skip_mypy: bool = Option(False, "--skip-mypy"),
    use_dmypy: bool = Option(
        False, "--dmypy", help="Use `dmypy run` instead of `mypy`"
    ),
    tool: str = ToolOption,
    dry: bool = DryOption,
) -> None:
    """Run: ruff check/format to reformat code and then mypy to check"""
    if getattr(files, "default", files) is None:
        files = ["."]
    elif isinstance(files, str):
        files = [files]
    skip = _ensure_bool(skip_mypy)
    dmypy = _ensure_bool(use_dmypy)
    bandit = _ensure_bool(bandit)
    prefix = _ensure_bool(prefix)
    tool = _ensure_str(tool)
    kwargs = {"dry": dry, "skip_mypy": skip, "dmypy": dmypy, "bandit": bandit}
    if _ensure_bool(check_only):
        check(files, tool=tool, **kwargs)
    else:
        lint(files, prefix=prefix, tool=tool, **kwargs)


@cli.command(name="check")
def only_check(
    bandit: bool = Option(False, "--bandit", help="Run `bandit -r <package_dir>`"),
    skip_mypy: bool = Option(False, "--skip-mypy"),
    dry: bool = DryOption,
) -> None:
    """Check code style without reformat"""
    check(dry=dry, bandit=bandit, skip_mypy=_ensure_bool(skip_mypy))


class Sync(DryRun):
    def __init__(
        self, filename: str, extras: str, save: bool, dry: bool = False
    ) -> None:
        self.filename = filename
        self.extras = extras
        self._save = save
        super().__init__(dry=dry)

    def gen(self) -> str:
        extras, save = self.extras, self._save
        should_remove = not Path.cwd().joinpath(self.filename).exists()
        if not (tool := Project.get_manage_tool()):
            if should_remove or not is_venv():
                raise EnvError("There project is not managed by uv/pdm/poetry!")
            return f"python -m pip install -r {self.filename}"
        prefix = "" if is_venv() else f"{tool} run "
        ensure_pip = " {1}python -m ensurepip && {1}python -m pip install -U pip &&"
        export_cmd = "uv export --no-hashes --all-extras --frozen"
        if tool in ("poetry", "pdm"):
            export_cmd = f"{tool} export --without-hashes --with=dev"
            if tool == "poetry":
                ensure_pip = ""
                if not UpgradeDependencies.should_with_dev():
                    export_cmd = export_cmd.replace(" --with=dev", "")
                if extras and isinstance(extras, (str, list)):
                    export_cmd += f" --{extras=}".replace("'", '"')
            elif check_call(prefix + "python -m pip --version"):
                ensure_pip = ""
        elif check_call(prefix + "python -m pip --version"):
            ensure_pip = ""
        install_cmd = (
            f"{{2}} -o {{0}} &&{ensure_pip} {{1}}python -m pip install -r {{0}}"
        )
        if should_remove and not save:
            install_cmd += " && rm -f {0}"
        return install_cmd.format(self.filename, prefix, export_cmd)


@cli.command()
def sync(
    filename: str = "dev_requirements.txt",
    extras: str = Option("", "--extras", "-E"),
    save: bool = Option(
        False, "--save", "-s", help="Whether save the requirement file"
    ),
    dry: bool = DryOption,
) -> None:
    """Export dependencies by poetry to a txt file then install by pip."""
    Sync(filename, extras, save, dry=dry).run()


def _should_run_test_script(path: Path = Path("scripts")) -> Path | None:
    for name in ("test.sh", "test.py"):
        if (file := path / name).exists():
            return file
    return None


def test(dry: bool, ignore_script: bool = False) -> None:
    cwd = Path.cwd()
    root = Project.get_work_dir(cwd=cwd, allow_cwd=True)
    script_dir = root / "scripts"
    if not _ensure_bool(ignore_script) and (
        test_script := _should_run_test_script(script_dir)
    ):
        cmd = f"{os.path.relpath(test_script, root)}"
        if cwd != root:
            cmd = f"cd {root} && " + cmd
    else:
        cmd = 'coverage run -m pytest -s && coverage report --omit="tests/*" -m'
        if not is_venv() or not check_call("coverage --version"):
            sep = " && "
            prefix = f"{tool} run " if (tool := Project.get_manage_tool()) else ""
            cmd = sep.join(prefix + i for i in cmd.split(sep))
    exit_if_run_failed(cmd, dry=dry)


@cli.command(name="test")
def coverage_test(
    dry: bool = DryOption,
    ignore_script: bool = Option(False, "--ignore-script", "-i"),
) -> None:
    """Run unittest by pytest and report coverage"""
    return test(dry, ignore_script)


class Publish:
    class CommandEnum(StrEnum):
        poetry = "poetry publish --build"
        pdm = "pdm publish"
        uv = "uv build && uv publish"
        twine = "python -m build && twine upload"

    @classmethod
    def gen(cls) -> str:
        if tool := Project.get_manage_tool():
            return cls.CommandEnum[tool]
        return cls.CommandEnum.twine


@cli.command()
def upload(
    dry: bool = DryOption,
) -> None:
    """Shortcut for package publish"""
    cmd = Publish.gen()
    exit_if_run_failed(cmd, dry=dry)


def dev(
    port: int | None | OptionInfo,
    host: str | None | OptionInfo,
    file: str | None | ArgumentInfo = None,
    dry: bool = False,
) -> None:
    cmd = "fastapi dev"
    no_port_yet = True
    if file is not None:
        try:
            port = int(str(file))
        except ValueError:
            cmd += f" {file}"
        else:
            if port != 8000:
                cmd += f" --port={port}"
                no_port_yet = False
    if no_port_yet and (port := getattr(port, "default", port)) and str(port) != "8000":
        cmd += f" --port={port}"
    if (host := getattr(host, "default", host)) and host not in (
        "localhost",
        "127.0.0.1",
    ):
        cmd += f" --host={host}"
    exit_if_run_failed(cmd, dry=dry)


@cli.command(name="dev")
def runserver(
    file_or_port: Optional[str] = typer.Argument(default=None),
    port: Optional[int] = Option(None, "-p", "--port"),
    host: Optional[str] = Option(None, "-h", "--host"),
    dry: bool = DryOption,
) -> None:
    """Start a fastapi server(only for fastapi>=0.111.0)"""
    if getattr(file_or_port, "default", file_or_port):
        dev(port, host, file=file_or_port, dry=dry)
    else:
        dev(port, host, dry=dry)


@cli.command(name="exec")
def run_by_subprocess(cmd: str, dry: bool = DryOption) -> None:
    """Run cmd by subprocess, auto set shell=True when cmd contains '|>'"""
    try:
        rc = run_and_echo(cmd, verbose=True, dry=_ensure_bool(dry))
    except FileNotFoundError as e:
        if e.filename == cmd.split()[0]:
            echo(f"Command not found: {e.filename}")
            raise Exit(1) from None
        raise e
    else:
        if rc:
            raise Exit(rc)


class MakeDeps(DryRun):
    def __init__(self, tool: str, prod: bool = False, dry: bool = False) -> None:
        self._tool = tool
        self._prod = prod
        super().__init__(dry=dry)

    def should_ensure_pip(self) -> bool:
        return True

    def should_upgrade_pip(self) -> bool:
        return True

    def get_groups(self) -> list[str]:
        if self._prod:
            return []
        return ["dev"]

    def gen(self) -> str:
        if self._tool == "pdm":
            return "pdm sync " + ("--prod" if self._prod else "-G :all")
        elif self._tool == "uv":
            return "uv sync --inexact --active" + (
                "" if self._prod else " --all-extras --all-groups"
            )
        elif self._tool == "poetry":
            return "poetry install " + (
                "--only=main" if self._prod else "--all-extras --all-groups"
            )
        else:
            cmd = "python -m pip install -e ."
            if gs := self.get_groups():
                cmd += " " + " ".join(f"--group {g}" for g in gs)
            upgrade = "python -m pip install --upgrade pip"
            if self.should_ensure_pip():
                cmd = f"python -m ensurepip && {upgrade} && {cmd}"
            elif self.should_upgrade_pip():
                cmd = "{upgrade} && {cmd}"
            return cmd


@cli.command(name="deps")
def make_deps(
    prod: bool = Option(
        False,
        "--prod",
        help="Only instead production dependencies.",
    ),
    tool: str = ToolOption,
    use_uv: bool = Option(False, "--uv", help="Use `uv` to install deps"),
    use_pdm: bool = Option(False, "--pdm", help="Use `pdm` to install deps"),
    use_pip: bool = Option(False, "--pip", help="Use `pip` to install deps"),
    use_poetry: bool = Option(False, "--poetry", help="Use `poetry` to install deps"),
    dry: bool = DryOption,
) -> None:
    """Run: ruff check/format to reformat code and then mypy to check"""
    if use_uv + use_pdm + use_pip + use_poetry > 1:
        raise UsageError("`--uv/--pdm/--pip/--poetry` can only choose one!")
    if use_uv:
        tool = "uv"
    elif use_pdm:
        tool = "pdm"
    elif use_pip:
        tool = "pip"
    elif use_poetry:
        tool = "poetry"
    elif tool == ToolOption.default:
        tool = Project.get_manage_tool(cache=True) or "pip"
    MakeDeps(tool, prod, dry=dry).run()


class UvPypi(DryRun):
    PYPI = "https://pypi.org/simple"
    HOST = "https://files.pythonhosted.org"

    def __init__(self, lock_file: Path, dry: bool, verbose: bool, quiet: bool) -> None:
        super().__init__(dry=dry)
        self.lock_file = lock_file
        self._verbose = _ensure_bool(verbose)
        self._quiet = _ensure_bool(quiet)

    def run(self) -> None:
        try:
            rc = self.update_lock(self.lock_file, self._verbose, self._quiet)
        except ValueError as e:
            secho(str(e), fg=typer.colors.RED)
            raise Exit(1) from e
        else:
            if rc != 0:
                raise Exit(rc)

    @classmethod
    def update_lock(cls, p: Path, verbose: bool, quiet: bool) -> int:
        text = p.read_text("utf-8")
        registry_pattern = r'(registry = ")(.*?)"'
        replace_registry = functools.partial(
            re.sub, registry_pattern, rf'\1{cls.PYPI}"'
        )
        registry_urls = {i[1] for i in re.findall(registry_pattern, text)}
        download_pattern = r'(url = ")(https?://.*?)(/packages/.*?\.)(gz|whl)"'
        replace_host = functools.partial(
            re.sub, download_pattern, rf'\1{cls.HOST}\3\4"'
        )
        download_hosts = {i[1] for i in re.findall(download_pattern, text)}
        if not registry_urls:
            raise ValueError(f"Failed to find pattern {registry_pattern!r} in {p}")
        if len(registry_urls) == 1:
            current_registry = registry_urls.pop()
            if current_registry == cls.PYPI:
                if download_hosts == {cls.HOST}:
                    if verbose:
                        echo(f"Registry of {p} is {cls.PYPI}, no need to change.")
                    return 0
            else:
                text = replace_registry(text)
                if verbose:
                    echo(f"{current_registry} --> {cls.PYPI}")
        else:
            # TODO: ask each one to confirm replace
            text = replace_registry(text)
            if verbose:
                for current_registry in sorted(registry_urls):
                    echo(f"{current_registry} --> {cls.PYPI}")
        if len(download_hosts) == 1:
            current_host = download_hosts.pop()
            if current_host != cls.HOST:
                text = replace_host(text)
                if verbose:
                    print(current_host, "-->", cls.HOST)
        elif download_hosts:
            # TODO: ask each one to confirm replace
            text = replace_host(text)
            if verbose:
                for current_host in sorted(download_hosts):
                    echo(f"{current_host} --> {cls.HOST}")
        size = p.write_text(text, encoding="utf-8")
        if verbose:
            echo(f"Updated {p} with {size} bytes.")
        if quiet:
            return 0
        return 1


@cli.command()
def pypi(
    file: Optional[str] = typer.Argument(default=None),
    dry: bool = DryOption,
    verbose: bool = False,
    quiet: bool = False,
) -> None:
    """Change registry of uv.lock to be pypi.org"""
    if not (p := Path(_ensure_str(file) or "uv.lock")).exists() and not (
        (p := Project.get_work_dir() / p.name).exists()
    ):
        yellow_warn(f"{p.name!r} not found!")
        return
    UvPypi(p, dry, verbose, quiet).run()


def version_callback(value: bool) -> None:
    if value:
        echo("Fast Dev Cli Version: " + typer.style(__version__, bold=True))
        raise Exit()


@cli.callback()
def common(
    version: bool = Option(
        None,
        "--version",
        "-V",
        callback=version_callback,
        is_eager=True,
        help="Show the version of this tool",
    ),
) -> None: ...


def main() -> None:
    cli()


if __name__ == "__main__":  # pragma: no cover
    main()

waketzheng / fast-dev-cli / 17573005466

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous