30735429562

Committed 02 Aug 2026 06:10AM UTC coverage: 81.947% (-0.6%) from 82.568%

Build # 30735429562

Build Type

push

github

Committed by

waketzheng

Commit Message

feat: auto load fastapi entrypoint for `fast dev --uvicorn`

Coverage Stats

2 of 20 new or added lines in 1 file covered. (10.0%)

1153 of 1407 relevant lines covered (81.95%)

4.09 hits per line

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

81.93

/fast_dev_cli/cli.py

from __future__ import annotations

import contextlib
import functools
import importlib.metadata as importlib_metadata
import os
import platform
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, Annotated, Any, Literal, cast, get_args, overload

import typer
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["uv", "pdm", "poetry"]
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:
        module_name = re.sub(r"[-_.]+", "-", name)
    else:
        module_name = canonicalize_name(name)
    return module_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


@functools.cache
def is_windows() -> bool:
    return platform.system() == "Windows"


@functools.cache
def prefer_uv_tool() -> bool:
    if shutil.which("uv") is None:
        return False
    cmd = "uv tool list"
    return Shell(cmd).capture_output() != "No tools installed"


def yellow_warn(msg: str) -> None:
    if is_windows() and (encoding := sys.stdout.encoding) != "utf-8":
        msg = msg.encode(encoding, errors="ignore").decode(encoding)
    secho(msg, fg="yellow")


def load_bool(name: str, default: bool = False, *, verbose: bool = True) -> bool:
    if not (v := os.getenv(name)):
        return default
    if (b := _convert_bool(v)) is not None:
        return b
    if verbose:
        secho(f"WARNING: can not convert value({v!r}) of {name} to bool!")
    return default


def _convert_bool(v: str) -> bool | None:
    if (lower := v.lower()) in ("0", "false", "f", "off", "no", "n"):
        return False
    elif lower in ("1", "true", "t", "on", "yes", "y"):
        return True
    return None


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
    )


class Shell:
    def __init__(self, cmd: list[str] | str, **kw: Any) -> None:
        self._cmd = cmd
        self._kw = kw

    @staticmethod
    def run_by_subprocess(
        cmd: list[str] | str, **kw: Any
    ) -> subprocess.CompletedProcess[str]:
        if isinstance(cmd, str):
            kw.setdefault("shell", True)
        check = kw.pop("check", False)
        return subprocess.run(cmd, check=check, **kw)  # nosec:B603

    @property
    def command(self) -> list[str] | str:
        command: list[str] | str = self._cmd
        if isinstance(command, str):
            cs = shlex.split(command)
            if "shell" not in self._kw and not (set(self._cmd) & {"|", ">", "&"}):
                command = self.expand_user(cs)
            elif any(i.startswith("~") for i in cs):
                command = re.sub(r" ~", " " + os.path.expanduser("~"), command)
        else:
            command = self.expand_user(command)
        return command

    @staticmethod
    def expand_user(cs: list[str]) -> list[str]:
        if cs[0] == "echo":
            return cs
        for i, c in enumerate(cs):
            if c.startswith("~"):
                cs[i] = os.path.expanduser(c)
        return cs

    def _run(self) -> subprocess.CompletedProcess[str]:
        return self.run_by_subprocess(self.command, **self._kw)

    def run(self, verbose: bool = False, dry: bool = False) -> int:
        if verbose:
            echo(f"--> {self._cmd}")
        if dry:
            return 0
        return self._run().returncode

    def check_call(self) -> bool:
        self._kw.update(stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        try:
            return self.run() == 0
        except FileNotFoundError:
            return False

    def capture_output(self, raises: bool = False) -> str:
        self._kw.update(capture_output=True, encoding="utf-8")
        r = self._run()
        if raises and r.returncode != 0:
            raise ShellCommandError(r.stderr)
        return (r.stdout or r.stderr or "").strip()

    def finish(
        self, env: dict[str, str] | None = None, _exit: bool = False, dry: bool = False
    ) -> subprocess.CompletedProcess[str]:
        self.run(verbose=True, dry=True)
        if _ensure_bool(dry):
            return subprocess.CompletedProcess("", 0)
        if env is not None:
            self._kw["env"] = {**os.environ, **env}
        r = self._run()
        if rc := r.returncode:
            if _exit:
                sys.exit(rc)
            raise Exit(rc)
        return r


def run_and_echo(
    cmd: str, *, dry: bool = False, verbose: bool = True, **kw: Any
) -> int:
    """Run shell command with subprocess and print it"""
    return Shell(cmd, **kw).run(verbose=verbose, dry=dry)


def check_call(cmd: str) -> bool:
    return Shell(cmd).check_call()


def capture_cmd_output(
    command: list[str] | str, *, raises: bool = False, **kw: Any
) -> str:
    return Shell(command, **kw).capture_output(raises=raises)


def exit_if_run_failed(
    cmd: str,
    env: dict[str, str] | None = None,
    _exit: bool = False,
    dry: bool = False,
    **kw: Any,
) -> subprocess.CompletedProcess[str]:
    return Shell(cmd, **kw).finish(env=env, _exit=_exit, dry=dry)


def _parse_version(line: str, pattern: re.Pattern[str]) -> str:
    return pattern.sub("", line).split("#")[0].strip().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*=")
    all_lines = Path(version_file).read_text("utf-8").strip().splitlines()
    for line in all_lines:
        if pattern.match(line):
            return _parse_version(line, pattern)
    pattern = re.compile(r"VERSION\s*=")
    for line in all_lines:
        if pattern.match(line):
            return _parse_version(line, pattern)
    secho(f"WARNING: can not find version pattern in {version_file}!")
    return "0.0.0"


def _get_frontend_version() -> tuple[Path, str] | None:
    try:
        frontend_version_file = Project.get_work_dir("package.json", be_file=True)
    except EnvError:
        return None
    try:
        from asynctor.jsons import json_loads
    except ImportError:
        from json import loads as json_loads  # type:ignore[assignment]
    content = frontend_version_file.read_bytes()
    metadata: dict[str, str] = json_loads(content)  # type:ignore
    try:
        current_version = metadata["version"]
    except (KeyError, TypeError):
        return None
    with contextlib.suppress(ValueError):
        frontend_version_file = frontend_version_file.relative_to(Path.cwd())
    return frontend_version_file, current_version


@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],
) -> 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():
        out = _get_poetry_project_version(verbose)
        if check_version:
            return True, out
        return out
    toml_text = work_dir = None
    if package_name is None:
        try:
            work_dir = Project.get_work_dir()
        except EnvError:
            if (res := _get_frontend_version()) is None:
                raise
            current_version = res[1]
            if check_version:
                return False, current_version
            return current_version
        else:
            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 _get_poetry_project_version(verbose: bool) -> str:
    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]
    return out


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


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


def _quote_shell_arg(value: str | Path) -> str:
    text = str(value)
    if not is_windows():
        return shlex.quote(text)
    if text and not re.search(r'[\s"&|<>^()%!]', text):
        return text

    # Quote for the Windows C runtime while keeping cmd.exe metacharacters inert.
    result = ['"']
    backslashes = 0
    for char in text:
        if char == "\\":
            backslashes += 1
        elif char == '"':
            result.append("\\" * (backslashes * 2 + 1))
            result.append(char)
            backslashes = 0
        else:
            result.append("\\" * backslashes)
            result.append(char)
            backslashes = 0
    result.append("\\" * (backslashes * 2))
    result.append('"')
    return "".join(result)


def _join_shell_args(args: list[str]) -> str:
    return " ".join(_quote_shell_arg(arg) for arg in args)


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:
            try:
                filename = self.parse_filename()
            except EnvError:
                if (res := _get_frontend_version()) is not None:
                    filename = res[0].name
                else:
                    raise
        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_dynamic_version(
        toml_text: str | None,
        context: dict[str, Any],
        work_dir: Path | None = None,
    ) -> str | None:
        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
        # e.g.: version = { source = "file", path = "fast_dev_cli/__init__.py" }
        v_key = "version = "
        p_key = 'path = "'
        if toml_text is None:
            toml_text = Project.load_toml_text()
        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
        return None

    @classmethod
    def parse_filename(
        cls,
        toml_text: str | None = None,
        work_dir: Path | None = None,
        package_name: str | None = None,
        context: dict[str, Any] | None = None,
    ) -> str:
        if context is None:
            if toml_text is None:
                toml_text = Project.load_toml_text()
            context = tomllib.loads(toml_text)
        is_dynamic_version = False
        with contextlib.suppress(KeyError):
            if "version" in context["project"]["dynamic"]:
                is_dynamic_version = True
        if is_dynamic_version:
            if filename := cls.parse_dynamic_version(toml_text, context, work_dir):
                return filename
            yellow_warn("Failed to find version file for this dynamic version project.")
        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() and (
                    filename := cls.parse_dynamic_version(toml_text, context, work_dir)
                ):
                    return filename
            else:
                by_version_plugin = version_value in ("0", "0.0.0", "init")
        if by_version_plugin:
            return cls.parse_plugin_version(context, package_name)
        return TOML_FILE

    @staticmethod
    def parse_plugin_version(context: dict[str, Any], package_name: str | None) -> str:
        try:
            package_item = context["tool"]["poetry"]["packages"]
        except KeyError:
            try:
                project_name = context["project"]["name"]
            except KeyError:
                packages: list[tuple[str, str]] = []
            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 pack_name, source_dir in packages:
            ds.append(cwd / pack_name)
            ds.append(cwd / "src" / pack_name)
            if source_dir and source_dir != "src":
                ds.append(cwd / source_dir / pack_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)

    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+)"'
        filename_arg = _quote_shell_arg(filename)
        cmd = (
            f'bumpversion {parse} --current-version="{_version}" {part} {filename_arg}'
        )
        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(only_me=True))
        ):
            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`")


def _echo_version(version_file: Any, value: str) -> None:
    styled = typer.style(value, bold=True)
    echo(f"Version value in {version_file}: " + styled)


@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):
        try:
            toml_text = Project.load_toml_text()
        except EnvError:
            if (res := _get_frontend_version()) is not None:
                _echo_version(*res)
                return
            raise
        doc = tomllib.loads(toml_text)
        if value := doc.get("project", {}).get("version", ""):
            styled = typer.style(value, bold=True, fg=typer.colors.CYAN)
            if project_name := doc["project"].get("name", ""):
                echo(f"{project_name} version: " + styled)
            else:
                echo(f"Got Version from {TOML_FILE}: " + styled)
            return
        version_file = doc["tool"]["pdm"]["version"]["path"]
        text = Project.get_work_dir().joinpath(version_file).read_text(encoding="utf-8")
        varname = "__version__"
        for line in text.splitlines():
            if line.strip().startswith(varname):
                value = line.split("=", 1)[-1].strip().strip('"').strip("'")
                _echo_version(version_file, value)
                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: bool | None = 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 get_poetry_version(command: str = "poetry") -> str:
        pattern = r"(\d+\.\d+\.\d+)"
        text = capture_cmd_output(f"{command} --version")
        for expr in (
            rf"Poetry \(version {pattern}\)",
            rf"Poetry.*version.*{pattern}.*\)",
            rf"{pattern}",
        ):
            if m := re.search(expr, text):
                return m.group(1)
        return ""

    @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
        backend = ""
        skip_uv = load_bool("FASTDEVCLI_SKIP_UV")
        with contextlib.suppress(KeyError, tomllib.TOMLDecodeError):
            doc = tomllib.loads(text)
            backend = doc["build-system"]["build-backend"]
            if skip_uv:
                for t in ("pdm", "poetry"):
                    if t in backend:
                        cls._tool = t
                        return cls._tool
        return cls._get_manage_tool(text, backend, skip_uv)

    @classmethod
    def _get_manage_tool(
        cls: type[Self], text: str, backend: str, skip_uv: bool
    ) -> ToolName | None:
        work_dir: Path | None = None
        uv_lock_exists: bool | None = None
        if skip_uv:
            for name in ("pdm", "poetry"):
                if f"[tool.{name}]" in text:
                    cls._tool = name
                    return cls._tool
            work_dir = cls.get_work_dir(allow_cwd=True)
            for name in ("pdm", "poetry"):
                if Path(work_dir, f"{name}.lock").exists():
                    cls._tool = cast(ToolName, name)
                    return cls._tool
            if uv_lock_exists := Path(work_dir, "uv.lock").exists():
                # Use pdm when uv is not available for uv managed project
                cls._tool = "pdm"
                return cls._tool
            return None
        if work_dir is None:
            work_dir = cls.get_work_dir(allow_cwd=True)
        if uv_lock_exists is None:
            uv_lock_exists = Path(work_dir, "uv.lock").exists()
        if uv_lock_exists:
            cls._tool = "uv"
            return cls._tool
        return cls._parse_manage_tool(work_dir, text, backend)

    @classmethod
    def _parse_manage_tool(
        cls: type[Self], work_dir: Path, text: str, backend: str
    ) -> ToolName | None:
        pdm_lock_exists = Path(work_dir, "pdm.lock").exists()
        poetry_lock_exists = Path(work_dir, "poetry.lock").exists()
        match pdm_lock_exists + poetry_lock_exists:
            case 1:
                cls._tool = "pdm" if pdm_lock_exists else "poetry"
                return cls._tool
            case _ as x:
                if backend:
                    for t in ("pdm", "poetry"):
                        if t in backend:
                            cls._tool = t
                            return cls._tool
                for name in ("pdm", "poetry"):
                    if f"[tool.{name}]" in text:
                        cls._tool = name
                        return cls._tool
                if x == 2:
                    cls._tool = (
                        "poetry" if load_bool("FASTDEVCLI_PREFER_POETRY") else "pdm"
                    )
                    return cls._tool
        if "[tool.uv]" in text or load_bool("FASTDEVCLI_PREFER_uv"):
            cls._tool = "uv"
            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[str, Any] | None = None, only_me: bool = False
    ) -> str:
        pdm_i = "pdm install --frozen" + " --prod" * prod
        if cls.is_pdm_project():
            return pdm_i
        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":
            install_me = "uv pip install -e ."
            if doc is None:
                doc = tomllib.loads(cls.load_toml_text())
            is_distribution = (
                doc.get("tool", {}).get("pdm", {}).get("distribution") is not False
            )
            if only_me:
                return install_me if is_distribution else pdm_i
            cmd = "uv sync --inexact" + " --no-dev" * prod
            if is_distribution:
                cmd += f" && {install_me}"
        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[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 install -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 "") 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 {_quote_shell_arg(_version)} "
            f"-m {_quote_shell_arg(self.message)} && 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 pdm publish")


@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,
        up: bool = False,
        sim: bool = True,
        strict: bool = False,
        ty: bool = False,
        fix: bool = True,
    ) -> 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
        self._up = up
        self._sim = sim
        self._strict = strict
        self._ty = _ensure_bool(ty)
        self._fix = _ensure_bool(fix)
        super().__init__(_exit, dry)

    @staticmethod
    def check_lint_tool_installed() -> bool:
        try:
            return check_call("ruff --version")
        except FileNotFoundError:
            # Windows may raise FileNotFoundError when ruff not installed
            return False

    @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 | list[str] = ".",
        check_only: bool = False,
        bandit: bool = False,
        skip_mypy: bool = False,
        use_dmypy: bool = False,
        tool: str = ToolOption.default,
        with_prefix: bool = False,
        ruff_check_up: bool = False,
        ruff_check_sim: bool = True,
        mypy_strict: bool = False,
        prefer_ty: bool = False,
        ruff_check_fix: bool = True,
    ) -> str:
        path_args = shlex.split(paths) if isinstance(paths, str) else paths
        if not path_args:
            path_args = ["."]
        quoted_paths = _join_shell_args(path_args)
        if path_args != ["."] and all(i.endswith(".html") for i in path_args):
            return f"prettier -w {quoted_paths}"
        ruff_rules = ["I", "B"]
        if ruff_check_sim and not load_bool("FASTDEVCLI_NO_SIM"):
            ruff_rules.append("SIM")
        if ruff_check_up or load_bool("FASTDEVCLI_UP"):
            ruff_rules.append("UP")
        ruff_check = "ruff check --extend-select=" + ",".join(ruff_rules)
        if (
            ruff_check_fix
            and not check_only
            and (not load_bool("NO_FIX") and not load_bool("FASTDEVCLI_NO_FIX"))
        ):
            ruff_check += " --fix"
        tools = ["ruff format", ruff_check, "mypy"]
        if check_only:
            tools[0] += " --check"
        if skip_mypy or load_bool("SKIP_MYPY") or load_bool("FASTDEVCLI_NO_MYPY"):
            # Sometimes mypy is too slow
            tools = tools[:-1]
        else:
            if prefer_ty or load_bool("FASTDEVCLI_TY"):
                tools[-1] = "ty check"
            else:
                if load_bool("IGNORE_MISSING_IMPORTS"):
                    tools[-1] += " --ignore-missing-imports"
                if mypy_strict or load_bool("FASTDEVCLI_STRICT"):
                    tools[-1] += " --strict"
        ruff_exists = cls.check_lint_tool_installed()
        prefix = ""
        should_run_by_tool = with_prefix
        requires_mypy = any(tool.startswith("mypy") for tool in tools)
        global_mypy = False
        if requires_mypy:
            # TODO: move this long logic to a single function
            local_bin = Path.home().joinpath(".local/bin")
            if local_bin.joinpath("mypy").exists():
                global_mypy = True
                mypy_opt = "--python-executable=.venv/bin/python"
                for i, t in enumerate(tools):
                    if t.startswith("mypy"):
                        if mypy_opt not in t:
                            tools[i] = t + " " + mypy_opt
                        break
            if not should_run_by_tool:
                if is_venv() and Path(sys.argv[0]).parent != local_bin:
                    # Virtual environment activated and fast-dev-cli is installed in it
                    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
                        elif prefer_uv_tool():
                            command = "uv tool install ruff"
                        yellow_warn(
                            "You may need to run the following command"
                            f" to install ruff:\n\n  {command}\n"
                        )
                    elif global_mypy:
                        should_run_by_tool = True
                    elif cls.missing_mypy_exec():
                        should_run_by_tool = True
                        if check_call('python -c "import fast_dev_cli"'):
                            command = "python -m pip install -U mypy"
                            yellow_warn(
                                "You may need to run the following command"
                                f" to install lint tools:\n\n  {command}\n"
                            )
                elif tool == ToolOption.default:
                    root = Project.get_work_dir(allow_cwd=True)
                    if py := shutil.which("python"):
                        try:
                            Path(py).relative_to(root)
                        except ValueError:
                            # Virtual environment not activated
                            should_run_by_tool = True
                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 = tool + " run "
                if tool == "uv":
                    if is_windows():
                        prefix += "--no-sync "
                    elif Path(bin_dir := ".venv/bin/").exists():
                        prefix = bin_dir
        if cls.prefer_dmypy(quoted_paths, tools, use_dmypy=use_dmypy):
            tools[-1] = "dmypy run"
        cmd = " && ".join(
            (
                tool
                # `ruff <command>` get the same result with `pdm run ruff <command>`.
                # Other tools should run inside the selected environment.
                if (
                    ruff_exists
                    and tool.startswith("ruff")
                    or (global_mypy and tool.startswith("mypy"))
                )
                else prefix + tool
            )
            + f" {quoted_paths}"
            for tool in 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 quoted_paths == "." and " -c " not in command:
                quoted_paths = _quote_shell_arg(cls.get_package_name())
            command += f" -r {quoted_paths}"
            cmd += " && " + command
        return cmd

    def gen(self) -> str:
        paths = ["."]
        if args := self.args:
            ps = shlex.split(args) if isinstance(args, str) else [str(i) for i in args]
            if len(ps) == 1:
                path = ps[0]
                if (
                    path != "."
                    # `Path("a.").suffix` got "." in py3.14 and got "" with py<3.14
                    and (p := Path(path)).suffix in ("", ".")
                    and not p.exists()
                ):
                    # e.g.:
                    # stem -> stem.py
                    # me. -> me.py
                    if path.endswith("."):
                        p = p.with_name(path[:-1])
                    for suffix in (".py", ".html"):
                        p = p.with_suffix(suffix)
                        if p.exists():
                            ps[0] = p.name
                            break
            paths = ps
        return self.to_cmd(
            paths,
            self.check_only,
            self._bandit,
            self._skip_mypy,
            self._use_dmypy,
            tool=self._tool,
            with_prefix=self._prefix,
            ruff_check_up=self._up,
            ruff_check_sim=self._sim,
            mypy_strict=self._strict,
            prefer_ty=self._ty,
            ruff_check_fix=self._fix,
        )


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,
    up: bool = False,
    sim: bool = True,
    strict: bool = False,
    ty: bool = False,
    fix: bool = True,
) -> 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,
        up=up,
        sim=sim,
        strict=strict,
        ty=ty,
        fix=fix,
    ).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,
    up: bool = False,
    sim: bool = True,
    strict: bool = False,
    ty: bool = False,
) -> None:
    LintCode(
        files,
        check_only=True,
        _exit=True,
        dry=dry,
        bandit=bandit,
        skip_mypy=skip_mypy,
        dmypy=dmypy,
        tool=tool,
        up=up,
        sim=sim,
        strict=strict,
        ty=ty,
    ).run()


def _should_bandit() -> bool:
    if v := os.getenv("FASTDEVCLI_BANDIT"):
        return _convert_bool(v) or True
    toml_text = Project.load_toml_text()
    lines = toml_text.splitlines()
    return any(i.startswith("[tool.bandit") for i in lines)


@cli.command(name="lint")
def make_style(
    files: list[str] | None = 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,
    up: bool = Option(False, help="Whether ruff check with --extend-select=UP"),
    sim: bool = Option(True, help="Whether ruff check with --extend-select=SIM"),
    strict: bool = Option(False, help="Whether run mypy with --strict"),
    ty: bool = Option(False, help="Whether use ty instead of mypy"),
    fix: bool | None = Option(None, help="Whether ruff check with --fix"),
    auto_bandit: bool | None = Option(
        None, help="Whether to run bandit if `[tool.bandit]` in pyproject.toml"
    ),
) -> 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) or (
        auto_bandit is not None and _ensure_bool(auto_bandit) and _should_bandit()
    )
    tool = _ensure_str(tool) or ""
    up = _ensure_bool(up)
    sim = _ensure_bool(sim)
    strict = _ensure_bool(strict)
    kwargs = {"dry": dry, "skip_mypy": skip, "dmypy": dmypy, "bandit": bandit}
    if _ensure_bool(check_only):
        run = check
    else:
        prefix = _ensure_bool(prefix)
        if fix is None or not isinstance(fix, bool):
            fix = load_bool("FASTDEVCLI_FIX", True)
        run = functools.partial(lint, prefix=prefix, fix=fix)
    run(files, tool=tool, up=up, sim=sim, strict=strict, ty=ty, **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,
    up: bool = Option(False, help="Whether ruff check with --extend-select=UP"),
    sim: bool = Option(True, help="Whether ruff check with --extend-select=SIM"),
    strict: bool = Option(False, help="Whether run mypy with --strict"),
    ty: bool = Option(False, help="Whether use ty instead of mypy"),
) -> None:
    """Check code style without reformat"""
    bandit = _ensure_bool(bandit)
    up = _ensure_bool(up)
    sim = _ensure_bool(sim)
    skip_mypy = _ensure_bool(skip_mypy)
    check(dry=dry, bandit=bandit, skip_mypy=skip_mypy, up=up, sim=sim, ty=ty)


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()
        filename = _quote_shell_arg(self.filename)
        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 {filename}"
        prefix = ""
        if not is_venv():
            prefix = f"{tool} run " + "--no-sync " * (tool == "uv")
        ensure_pip = " {1}python -m ensurepip && {1}python -m pip install -U pip &&"
        export_cmd = "uv export --no-hashes --all-extras --all-groups --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={_quote_shell_arg(str(extras))}"
            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(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 = _quote_shell_arg(test_script.relative_to(root).as_posix())
        if test_script.suffix == ".py":
            cmd = "python " + cmd
        if cwd != root:
            cmd = f"cd {_quote_shell_arg(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, tool: str | None, verbose: bool) -> str:
        if tool == "auto":
            tool = Project.get_manage_tool() or ""
        if tool:
            if tool == "uv":
                envs = ["UV_PUBLISH_INDEX", "UV_PUBLISH_URL", "UV_PUBLISH_TOKEN"]
                if not any(os.getenv(i) for i in envs):
                    if verbose:
                        yellow_warn(f"Skip uv publish as envs ({envs}) not set")
                    return "uv build"
            elif tool == "pdm":
                env = "PDM_PUBLISH_REPO"
                if not os.getenv(env):
                    if verbose:
                        yellow_warn(f"Skip pdm publish as env ({env}) not set")
                    return "pdm build"
            return cls.CommandEnum[tool]
        return cls.CommandEnum.twine


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


def should_use_just() -> bool:
    if shutil.which("just") is None or load_bool("FASTDEVCLI_IGNORE_JUST_DEV"):
        return False
    d = Path.cwd()
    for _ in range(5):
        f = d / "justfile"
        if f.exists():
            return _prefer_just_dev(f)
        if d.joinpath("pyproject.toml").exists():
            break
        d = d.parent
    return False


def _prefer_just_dev(f: Path) -> bool:
    text = f.read_text(encoding="utf-8")
    lines = text.splitlines()
    dev_recipe = "dev *args:"
    re_import = re.compile(r"import[?]? ")
    has_import = False
    total = len(lines)
    for i, line in enumerate(lines):
        if line.startswith(dev_recipe):
            # Avoid cycle callback
            command_lines = []
            for j in range(i + 1, total):
                try:
                    s = lines[j]
                except IndexError:
                    break
                if not s.startswith(" "):
                    break
                command_lines.append(s)
            if not command_lines:  # Invalid justfile
                return False
            return all("fast dev" not in i for i in command_lines)
        elif not has_import and re_import.match(line):
            has_import = True
    if has_import:
        recipes = capture_cmd_output("just --list").splitlines()
        for recipe in recipes:
            recipe = recipe.strip()
            if recipe.startswith(dev_recipe):
                return "fast dev" not in recipe
    return False


def _load_fastapi_entrypoint() -> str:
    with contextlib.suppress(FileNotFoundError, KeyError):
        try:
            toml_text = Project.load_toml_text()
        except EnvError:
            return ""
        doc = tomllib.loads(toml_text)
        return doc["tool"]["fastapi"]["entrypoint"]
    return ""


def _parse_serve_file(
    uvicorn: bool | None, filename: str, cmd: str, args: list[str]
) -> str:
    if m := re.search(r"(.*):(\d+)$", filename):
        h, p = m.group(1), m.group(2)
        if h and "--host" not in str(args):
            if h == "0":
                args.append("--host=0.0.0.0")
            else:
                args.append(f"--host={h}")
        args.append(f"--port={p}")
        if uvicorn:
            if entrypoint := _load_fastapi_entrypoint():
                cmd += " " + entrypoint
            else:
                p = Path("main.py")
                if p.exists():
                    cmd += " main:app"
                elif Path("app", p.name).exists():
                    cmd += " app.main:app"
                elif Path("app.py").exists():
                    cmd += " app:app"
        return cmd
    if uvicorn and ((filepath := Path(filename)).is_file() or filepath.suffix == ".py"):
        filename = filepath.stem + ":app"
        parent_names = [j for i in filepath.parents if (j := i.name)]
        if parent_names:
            filename = ".".join([*parent_names[::-1], filename])
    cmd += " " + _quote_shell_arg(filename)
    return cmd


def _runserver(
    uvicorn: bool | None,
    host: OptionInfo | str | None,
    port: OptionInfo | int | None,
    file: ArgumentInfo | str | None,
) -> tuple[str, list[str]]:
    cmd = "uvicorn" if uvicorn else "fastapi dev"
    args = []
    if (host := getattr(host, "default", host)) and host not in (
        "localhost",
        "127.0.0.1",
    ):
        args.append(f"--host={host}")
    no_port_yet = True
    if file is not None:
        filename = str(file)
        try:
            port = int(filename)
        except ValueError:
            cmd = _parse_serve_file(uvicorn, filename, cmd, args)
        else:
            if port != 8000:
                args.append(f"--port={port}")
                no_port_yet = False
    elif uvicorn and (entrypoint := _load_fastapi_entrypoint()):
        cmd += " " + entrypoint
    if no_port_yet and (port := getattr(port, "default", port)) and str(port) != "8000":
        args.append(f"--port={port}")
    if shutil.which("pdm") is not None:
        cmd = "pdm run " + cmd
    return cmd, args


def dev(
    port: int | None | OptionInfo,
    host: str | None | OptionInfo,
    fastapi: bool | None = None,
    uvicorn: bool | None = None,
    prod: bool | None = None,
    reload: bool | None = None,
    just: bool | None = None,
    file: str | None | ArgumentInfo = None,
    dry: bool = False,
) -> None:
    if just is not False and should_use_just():
        args = [i for i in sys.argv[2:] if i != "--dry"]
        cmd = "just dev"
    else:
        cmd, args = _runserver(uvicorn, host, port, file)
    if args:
        cmd += " " + _join_shell_args(args)
    exit_if_run_failed(cmd, dry=dry)


@cli.command(name="dev")
def runserver(
    file_or_port: str | None = typer.Argument(default=None),
    port: int | None = Option(None, "-p", "--port"),
    host: str | None = Option(None, "-h", "--host"),
    fastapi: bool | None = None,
    uvicorn: bool | None = None,
    prod: bool | None = None,
    reload: bool | None = None,
    just: bool | None = None,
    dry: bool = DryOption,
) -> None:
    """Start a fastapi server(only for fastapi>=0.111.0)"""
    f = functools.partial(
        dev, port, host, fastapi, uvicorn, prod, reload, just, dry=dry
    )
    if getattr(file_or_port, "default", file_or_port):
        f(file=file_or_port)
    else:
        f()


@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:
        command = cmd.split()[0]
        if e.filename == command or (
            e.filename is None and "系统找不到指定的文件" in str(e)
        ):
            echo(f"Command not found: {command}")
            raise Exit(1) from None
        raise
    else:
        if rc:
            raise Exit(rc)


class MakeDeps(DryRun):
    def __init__(
        self,
        tool: str,
        prod: bool = False,
        dry: bool = False,
        active: bool = False,
        inexact: bool = False,
        no_dev: bool = False,
        verbose: bool = False,
        frozen: bool = False,
        no_extra: list[str] | None = None,
        no_group: list[str] | None = None,
    ) -> None:
        self._tool = tool
        self._prod = prod
        self._active = active or load_bool("FASTDEVCLI_DEPS_ACTIVE")
        self._inexact = inexact or load_bool("FASTDEVCLI_DEPS_INEXACT")
        self._verbose = verbose
        self._frozen = frozen
        self._no_dev = no_dev
        self._no_extra = no_extra
        self._no_group = no_group
        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:
        cmd = self._gen()
        if self._verbose:
            cmd += " --verbose"
        if self._no_dev:
            opt = " --no-dev"
            if opt not in cmd:
                cmd += opt
        if self._no_extra:
            cmd += " " + " ".join(
                f"--no-extra {_quote_shell_arg(i)}" for i in self._no_extra
            )
        if self._no_group:
            cmd += " " + " ".join(
                f"--no-group {_quote_shell_arg(i)}" for i in self._no_group
            )
        if self._frozen:
            cmd += " --frozen"
        if opts := os.getenv("FASTDEVCLI_DEPS_OPTS"):
            cmd += " " + opts.strip()
        return cmd

    def get_package_name(self) -> str:
        with contextlib.suppress(FileNotFoundError, KeyError):
            try:
                toml_text = Project.load_toml_text()
            except EnvError:
                return ""
            doc = tomllib.loads(toml_text)
            tool_section = doc["tool"]
            uv_package = tool_section.get("uv", {}).get("package")
            if uv_package is not None:
                if not uv_package:
                    return ""
            else:
                match doc["build-system"]["build-backend"]:
                    case "pdm.backend":
                        if not tool_section.get("pdm", {}).get("distribution", True):
                            return ""
                    case x if x.startswith("poetry"):
                        if not tool_section.get("poetry", {}).get("package-mode", True):
                            return ""
            return cast(str, doc["project"]["name"])
        return ""

    def _gen(self) -> str:
        if self._tool == "pdm":
            return "pdm install --frozen " + ("--prod" if self._prod else "-G :all")
        elif self._tool == "uv":
            uv_sync = "uv sync"
            if project := self.get_package_name():
                uv_sync += " " + _quote_shell_arg(f"--reinstall-package={project}")
            uv_sync += " --inexact" * self._inexact + " --active" * self._active
            return uv_sync + (
                " --no-dev" 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 = f"{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"),
    active: bool = Option(
        False, help="Add `--active` to uv sync command(Only work for uv project)"
    ),
    inexact: bool = Option(
        False, help="Add `--inexact` to uv sync command(Only work for uv project)"
    ),
    no_dev: bool = Option(False, "--no-dev"),
    no_extra: Annotated[list[str] | None, Option()] = None,
    no_group: Annotated[list[str] | None, Option()] = None,
    frozen: bool = Option(False, "--frozen", "--frozen-lockfile", "--no-lock"),
    verbose: bool = Option(False, "--verbose"),
    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 typer.BadParameter(
            "can only choose one",
            param_hint=("--uv", "--pdm", "--pip", "--poetry"),
        )
    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"
    bool_opts = {
        "active": active,
        "inexact": inexact,
        "no_dev": no_dev,
        "verbose": verbose,
        "frozen": frozen,
        "dry": dry,
    }
    MakeDeps(tool, prod, no_extra=no_extra, no_group=no_group, **bool_opts).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,
        slim: bool = False,
        reverse: bool = False,
    ) -> None:
        super().__init__(dry=dry)
        self.lock_file = lock_file
        self._verbose = _ensure_bool(verbose)
        self._quiet = _ensure_bool(quiet)
        self._slim = _ensure_bool(slim)
        self._reverse = _ensure_bool(reverse)

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

    @staticmethod
    def get_target_content(
        text: str, verbose: bool, target_registry: str, target_host: str
    ) -> str | None:
        registry_pattern = r'(registry = ")(.*?)"'
        registry_urls = {i[1] for i in re.findall(registry_pattern, text)}
        download_pattern = r'(url = ")(https?://.*?)(/packages/.*?\.)(gz|whl|zip)"'
        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 uv lock file"
            )

        def replace_registry(s: str) -> str:
            return re.sub(registry_pattern, rf'\1{target_registry}"', s)

        def replace_host(s: str) -> str:
            return re.sub(download_pattern, rf'\1{target_host}\3\4"', s)

        if len(registry_urls) == 1:
            current_registry = registry_urls.pop()
            if current_registry == target_registry:
                if download_hosts == {target_host}:
                    return None
            else:
                text = replace_registry(text)
                if verbose:
                    echo(f"{current_registry} --> {target_registry}")
        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} --> {target_registry}")
        if len(download_hosts) == 1:
            current_host = download_hosts.pop()
            if current_host != target_host:
                text = replace_host(text)
                if verbose:
                    echo(f"{current_host} --> {target_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} --> {target_host}")
        return text

    @classmethod
    def update_lock(
        cls,
        p: Path,
        verbose: bool,
        quiet: bool,
        slim: bool = False,
        reverse: bool = False,
    ) -> int:
        text = p.read_text("utf-8")
        target_register, target_host = cls.PYPI, cls.HOST
        if reverse:
            try:
                target_register, target_host = cls.get_register_from_uv_config()
            except FileNotFoundError:
                if verbose:
                    echo("Skip register reverse as global uv config file not found.")
                if quiet:
                    return 0
                return 1
        new_text = cls.get_target_content(text, verbose, target_register, target_host)
        if new_text is None:
            if verbose:
                echo(f"Registry of {p} is {target_register}, no need to change.")
            return 0
        return cls.slim_and_write(new_text, slim, p, verbose, quiet)

    @classmethod
    def get_register_from_uv_config(cls) -> tuple[str, str]:
        config_file = cls.get_uv_config_file()
        text = config_file.read_text("utf-8")
        doc = tomllib.loads(text)
        index_url = doc["index"][0]["url"]
        return index_url, index_url.replace("/simple", "").rstrip("/")

    @staticmethod
    def get_uv_config_file() -> Path:
        config_dir = "AppData/Roaming" if is_windows() else ".config"
        return Path.home() / config_dir / "uv/uv.toml"

    @staticmethod
    def slim_and_write(
        text: str, slim: bool, p: Path, verbose: bool, quiet: bool
    ) -> int:
        if slim:
            pattern = r', size = \d+, upload-time = ".*?"'
            text = re.sub(pattern, "", text)
        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: str | None = typer.Argument(default=None),
    dry: bool = DryOption,
    verbose: bool = False,
    quiet: bool = False,
    slim: bool = False,
    reverse: 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, slim, reverse).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 / 30735429562

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