feat: 兼容windows

main
NoahLan 2 weeks ago
parent 8e23edcb90
commit bab809daf8

@ -11,12 +11,14 @@
```bash ```bash
uv tool install git+https://git.noahlan.cn/iti-framework/iti-flask-cli.git uv tool install git+https://git.noahlan.cn/iti-framework/iti-flask-cli.git
uv tool update-shell
``` ```
或: 或:
```bash ```bash
pipx install git+https://git.noahlan.cn/iti-framework/iti-flask-cli.git pipx install git+https://git.noahlan.cn/iti-framework/iti-flask-cli.git
pipx ensurepath
``` ```
本地开发安装: 本地开发安装:
@ -24,8 +26,11 @@ pipx install git+https://git.noahlan.cn/iti-framework/iti-flask-cli.git
```bash ```bash
cd iti-flask-cli cd iti-flask-cli
uv tool install -e . uv tool install -e .
uv tool update-shell
``` ```
Windows 安装后重新打开终端,再执行 `iticli help`
## 项目识别 ## 项目识别
`iticli` 会从当前目录向上查找 `pyproject.toml`,并识别两类项目: `iticli` 会从当前目录向上查找 `pyproject.toml`,并识别两类项目:
@ -51,6 +56,7 @@ iticli migrate
iticli migrate heads iticli migrate heads
iticli migrate current iticli migrate current
iticli migrate revision "alice add order table" iticli migrate revision "alice add order table"
iticli migrate --env prod
``` ```
业务项目: 业务项目:

@ -22,6 +22,12 @@ DEFAULT_SYSTEM_GIT = os.environ.get(
DEFAULT_COPIER_REF = os.environ.get("ITICLI_COPIER_REF", "HEAD") DEFAULT_COPIER_REF = os.environ.get("ITICLI_COPIER_REF", "HEAD")
ENV_NAMES = {"dev", "test", "prod", "default"} ENV_NAMES = {"dev", "test", "prod", "default"}
COMMAND_HINTS = {
"uv": "Install uv and run `uv tool update-shell`, then reopen the terminal.",
"docker": "Install Docker Desktop and make sure `docker` is on PATH.",
"git": "Install Git and make sure `git` is on PATH.",
"pipx": "Install pipx and make sure `pipx` is on PATH.",
}
class CliError(Exception): class CliError(Exception):
@ -76,6 +82,7 @@ def build_parser() -> argparse.ArgumentParser:
add_run_parser(subparsers, "serve") add_run_parser(subparsers, "serve")
migrate_parser = subparsers.add_parser("migrate", help="run Alembic commands") migrate_parser = subparsers.add_parser("migrate", help="run Alembic commands")
migrate_parser.add_argument("--env", dest="env_name", choices=sorted(ENV_NAMES), help="APP_ENV for this migration")
migrate_parser.add_argument("action", nargs="?", help="upgrade target, heads, current, revision, ...") migrate_parser.add_argument("action", nargs="?", help="upgrade target, heads, current, revision, ...")
migrate_parser.add_argument("args", nargs=argparse.REMAINDER) migrate_parser.add_argument("args", nargs=argparse.REMAINDER)
migrate_parser.add_argument("-m", "--message", help="migration message for revision") migrate_parser.add_argument("-m", "--message", help="migration message for revision")
@ -185,24 +192,26 @@ def handle_migrate(args: argparse.Namespace) -> int:
action = args.action action = args.action
rest = list(args.args or []) rest = list(args.args or [])
base_cmd = alembic_base_cmd(ctx) base_cmd = alembic_base_cmd(ctx)
env_name = normalize_env_name(args.env_name) if args.env_name else extract_env_name(rest)
extra_env = {"APP_ENV": env_name} if env_name else None
if not action: if not action:
return run([*base_cmd, "upgrade", "head"], ctx.root) return run([*base_cmd, "upgrade", "head"], ctx.root, extra_env=extra_env)
if action in {"revision", "migration"}: if action in {"revision", "migration"}:
message = args.message or extract_message(rest) or " ".join(rest).strip() message = args.message or extract_message(rest) or " ".join(rest).strip()
if not message: if not message:
raise CliError('missing migration message, example: iticli migrate revision "alice add order table"') raise CliError('missing migration message, example: iticli migrate revision "alice add order table"')
return run([*base_cmd, "revision", "--autogenerate", "-m", message], ctx.root) return run([*base_cmd, "revision", "--autogenerate", "-m", message], ctx.root, extra_env=extra_env)
if action == "upgrade": if action == "upgrade":
target = rest[0] if rest else "head" target = rest[0] if rest else "head"
return run([*base_cmd, "upgrade", target, *rest[1:]], ctx.root) return run([*base_cmd, "upgrade", target, *rest[1:]], ctx.root, extra_env=extra_env)
if action in {"heads", "current", "history", "branches", "downgrade", "stamp", "show"}: if action in {"heads", "current", "history", "branches", "downgrade", "stamp", "show"}:
return run([*base_cmd, action, *rest], ctx.root) return run([*base_cmd, action, *rest], ctx.root, extra_env=extra_env)
return run([*base_cmd, "upgrade", action, *rest], ctx.root) return run([*base_cmd, "upgrade", action, *rest], ctx.root, extra_env=extra_env)
def handle_create(args: argparse.Namespace) -> int: def handle_create(args: argparse.Namespace) -> int:
@ -220,7 +229,9 @@ def handle_create(args: argparse.Namespace) -> int:
display_name = args.display_name or target.name display_name = args.display_name or target.name
cmd = [ cmd = [
"uvx", "uv",
"tool",
"run",
"copier", "copier",
"copy", "copy",
"--defaults", "--defaults",
@ -306,14 +317,15 @@ def update_sync_cmd(ctx: ProjectContext, updates: Mapping[str, str]) -> list[str
def handle_template(args: argparse.Namespace) -> int: def handle_template(args: argparse.Namespace) -> int:
ctx = require_project({"business"}) ctx = require_project({"business"})
cmd = ["uvx", "copier", "update", "--defaults", "--vcs-ref", "HEAD", str(ctx.root)] cmd = ["uv", "tool", "run", "copier", "update", "--defaults", "--vcs-ref", "HEAD", str(ctx.root)]
if args.template_action == "check": if args.template_action == "check":
cmd.insert(4, "--pretend") cmd.insert(6, "--pretend")
return run(cmd, ctx.root) return run(cmd, ctx.root)
def handle_init(args: argparse.Namespace) -> int: def handle_init(args: argparse.Namespace) -> int:
ctx = require_project({"business"}) ctx = require_project({"business"})
ensure_env_file(ctx.root)
if args.scope == "system": if args.scope == "system":
if not ctx.has_system: if not ctx.has_system:
raise CliError("current business project does not depend on iti-system") raise CliError("current business project does not depend on iti-system")
@ -342,6 +354,7 @@ def handle_init(args: argparse.Namespace) -> int:
def handle_docker(args: argparse.Namespace) -> int: def handle_docker(args: argparse.Namespace) -> int:
ctx = require_project({"business"}) ctx = require_project({"business"})
ensure_env_file(ctx.root)
compose = ["docker", "compose"] compose = ["docker", "compose"]
if args.db: if args.db:
compose.extend(["-f", "docker-compose.yml", "-f", "docker-compose.with-db.yml"]) compose.extend(["-f", "docker-compose.yml", "-f", "docker-compose.with-db.yml"])
@ -454,7 +467,7 @@ def parse_run_args(env_or_port: str | None, port_arg: str | None) -> tuple[str,
if env_or_port: if env_or_port:
if env_or_port in ENV_NAMES: if env_or_port in ENV_NAMES:
env_name = "dev" if env_or_port == "default" else env_or_port env_name = normalize_env_name(env_or_port)
port = port_arg or port port = port_arg or port
elif env_or_port.isdigit(): elif env_or_port.isdigit():
port = env_or_port port = env_or_port
@ -469,22 +482,64 @@ def parse_run_args(env_or_port: str | None, port_arg: str | None) -> tuple[str,
return env_name, str(port) return env_name, str(port)
def normalize_env_name(raw: str) -> str:
if raw not in ENV_NAMES:
supported = ", ".join(sorted(ENV_NAMES))
raise CliError(f"invalid APP_ENV: {raw}; expected one of {supported}")
return "dev" if raw == "default" else raw
def extract_env_name(rest: list[str]) -> str | None:
value = extract_remainder_option(rest, "--env")
return normalize_env_name(value) if value else None
def extract_message(rest: list[str]) -> str | None: def extract_message(rest: list[str]) -> str | None:
for flag in ("-m", "--message"): return extract_remainder_option(rest, "-m", "--message")
def extract_remainder_option(rest: list[str], *flags: str) -> str | None:
for flag in flags:
if flag in rest: if flag in rest:
index = rest.index(flag) index = rest.index(flag)
if index + 1 >= len(rest): if index + 1 >= len(rest):
raise CliError(f"{flag} requires a message") raise CliError(f"{flag} requires a value")
message = rest[index + 1] value = rest[index + 1]
del rest[index : index + 2] del rest[index : index + 2]
return message return value
for item in list(rest): for item in list(rest):
if item.startswith("--message="): for flag in flags:
rest.remove(item) if item.startswith(f"{flag}="):
return item.partition("=")[2] rest.remove(item)
return item.partition("=")[2]
return None return None
def resolve_executable(command: str) -> str:
if not command:
raise CliError("empty command")
if is_path_command(command):
return command
resolved = shutil.which(command)
if resolved:
return resolved
raise CliError(missing_command_message(command))
def is_path_command(command: str) -> bool:
if Path(command).is_absolute():
return True
return any(separator and separator in command for separator in (os.sep, os.altsep))
def missing_command_message(command: str) -> str:
hint = COMMAND_HINTS.get(command)
if hint:
return f"command not found: {command}. {hint}"
return f"command not found: {command}"
def alembic_base_cmd(ctx: ProjectContext) -> list[str]: def alembic_base_cmd(ctx: ProjectContext) -> list[str]:
cmd = ["uv", "run", "python", "-m", "alembic"] cmd = ["uv", "run", "python", "-m", "alembic"]
if ctx.kind == "business": if ctx.kind == "business":
@ -500,10 +555,18 @@ def seed_system(ctx: ProjectContext) -> int:
return run( return run(
["uv", "run", "iti-system", "seed", "system", "app:create_app"], ["uv", "run", "iti-system", "seed", "system", "app:create_app"],
ctx.root, ctx.root,
extra_env={"PYTHONPATH": "."}, extra_env={"PYTHONPATH": str(ctx.root)},
) )
def ensure_env_file(root: Path) -> None:
env_file = root / ".env"
env_example = root / ".env.example"
if not env_file.exists() and env_example.is_file():
shutil.copyfile(env_example, env_file)
print("created .env from .env.example")
def update_self() -> int: def update_self() -> int:
commands: list[list[str]] = [] commands: list[list[str]] = []
if shutil.which("uv"): if shutil.which("uv"):
@ -522,7 +585,10 @@ def update_self() -> int:
def run(cmd: Sequence[str], cwd: Path, extra_env: dict[str, str] | None = None) -> int: def run(cmd: Sequence[str], cwd: Path, extra_env: dict[str, str] | None = None) -> int:
env = project_env(cwd, extra_env) env = project_env(cwd, extra_env)
completed = subprocess.run(list(cmd), cwd=str(cwd), env=env, check=False) if not cmd:
raise CliError("empty command")
resolved_cmd = [resolve_executable(str(cmd[0])), *[str(item) for item in cmd[1:]]]
completed = subprocess.run(resolved_cmd, cwd=str(cwd), env=env, check=False)
return int(completed.returncode) return int(completed.returncode)
@ -538,10 +604,21 @@ def project_env(root: Path, extra_env: dict[str, str] | None = None) -> dict[str
if not same_venv: if not same_venv:
env.pop("VIRTUAL_ENV", None) env.pop("VIRTUAL_ENV", None)
if extra_env: if extra_env:
env.update(extra_env) for key, value in extra_env.items():
if key == "PYTHONPATH":
env[key] = merge_path_env(str(value), env.get(key))
else:
env[key] = value
return env return env
def merge_path_env(value: str, current: str | None) -> str:
parts = [part for part in value.split(os.pathsep) if part]
if current:
parts.extend(part for part in current.split(os.pathsep) if part)
return os.pathsep.join(parts)
def release_framework(root: Path, version_arg: str | None) -> int: def release_framework(root: Path, version_arg: str | None) -> int:
current_version = read_current_version(root / "iti" / "__about__.py") current_version = read_current_version(root / "iti" / "__about__.py")
target_version = normalize_version(version_arg) if version_arg else bump_patch(current_version) target_version = normalize_version(version_arg) if version_arg else bump_patch(current_version)
@ -557,7 +634,7 @@ def release_framework(root: Path, version_arg: str | None) -> int:
branch = git_output(root, ["symbolic-ref", "--quiet", "--short", "HEAD"]).strip() branch = git_output(root, ["symbolic-ref", "--quiet", "--short", "HEAD"]).strip()
if not branch: if not branch:
raise CliError("release requires a branch checkout") raise CliError("release requires a branch checkout")
if subprocess.run(["git", "rev-parse", "--verify", "--quiet", f"refs/tags/{target_tag}"], cwd=root).returncode == 0: if git_ref_exists(root, f"refs/tags/{target_tag}"):
raise CliError(f"tag already exists: {target_tag}") raise CliError(f"tag already exists: {target_tag}")
code = run(["uv", "run", "pytest", "-q"], root) code = run(["uv", "run", "pytest", "-q"], root)
@ -608,7 +685,7 @@ def release_system(root: Path, version_arg: str | None) -> int:
branch = git_output(root, ["symbolic-ref", "--quiet", "--short", "HEAD"]).strip() branch = git_output(root, ["symbolic-ref", "--quiet", "--short", "HEAD"]).strip()
if not branch: if not branch:
raise CliError("release requires a branch checkout") raise CliError("release requires a branch checkout")
if subprocess.run(["git", "rev-parse", "--verify", "--quiet", f"refs/tags/{target_tag}"], cwd=root).returncode == 0: if git_ref_exists(root, f"refs/tags/{target_tag}"):
raise CliError(f"tag already exists: {target_tag}") raise CliError(f"tag already exists: {target_tag}")
code = run(["uv", "run", "pytest", "-q"], root) code = run(["uv", "run", "pytest", "-q"], root)
@ -634,10 +711,21 @@ def release_system(root: Path, version_arg: str | None) -> int:
return 0 return 0
def git_ref_exists(root: Path, ref: str) -> bool:
completed = subprocess.run(
[resolve_executable("git"), "rev-parse", "--verify", "--quiet", ref],
cwd=str(root),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
return completed.returncode == 0
def git_output(root: Path, args: Sequence[str]) -> str: def git_output(root: Path, args: Sequence[str]) -> str:
completed = subprocess.run( completed = subprocess.run(
["git", *args], [resolve_executable("git"), *args],
cwd=root, cwd=str(root),
text=True, text=True,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
@ -650,7 +738,7 @@ def git_output(root: Path, args: Sequence[str]) -> str:
def latest_git_tag(git_url: str) -> str: def latest_git_tag(git_url: str) -> str:
completed = subprocess.run( completed = subprocess.run(
["git", "ls-remote", "--tags", git_url], [resolve_executable("git"), "ls-remote", "--tags", git_url],
text=True, text=True,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,

@ -1,23 +1,31 @@
from __future__ import annotations from __future__ import annotations
import os
from pathlib import Path from pathlib import Path
import pytest import pytest
from iticli import cli as cli_module
from iticli.cli import ( from iticli.cli import (
CliError,
ProjectContext, ProjectContext,
alembic_base_cmd, alembic_base_cmd,
build_run_cmd, build_run_cmd,
build_parser, build_parser,
detect_project, detect_project,
ensure_env_file,
ensure_uv_no_sources_package, ensure_uv_no_sources_package,
extract_env_name,
extract_message, extract_message,
find_optional_extra, find_optional_extra,
git_tags_from_refs, git_tags_from_refs,
has_package, has_package,
latest_git_tag_from_refs, latest_git_tag_from_refs,
merge_path_env,
normalize_version, normalize_version,
parse_run_args, parse_run_args,
project_env,
resolve_executable,
resolve_update_packages, resolve_update_packages,
update_sync_cmd, update_sync_cmd,
update_pyproject_git_tags, update_pyproject_git_tags,
@ -163,6 +171,13 @@ def test_extract_message_accepts_remainder_flags() -> None:
assert rest == ["--head", "head"] assert rest == ["--head", "head"]
def test_extract_env_name_accepts_remainder_flags() -> None:
rest = ["--env", "prod", "--head", "head"]
assert extract_env_name(rest) == "prod"
assert rest == ["--head", "head"]
def test_parser_exposes_update_instead_of_sync() -> None: def test_parser_exposes_update_instead_of_sync() -> None:
parser = build_parser() parser = build_parser()
@ -185,6 +200,121 @@ def test_parser_accepts_create_database_dialect() -> None:
assert args.database == "postgresql" assert args.database == "postgresql"
def test_parser_accepts_migrate_env() -> None:
parser = build_parser()
args = parser.parse_args(["migrate", "--env", "prod"])
assert args.command == "migrate"
assert args.env_name == "prod"
def test_create_uses_uv_tool_run_copier(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
calls = []
def fake_run(cmd, cwd, extra_env=None):
calls.append((cmd, cwd, extra_env))
return 0
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(cli_module, "run", fake_run)
args = build_parser().parse_args(["create", "--database", "postgresql", "demo"])
assert args.handler(args) == 0
cmd, cwd, extra_env = calls[0]
assert cmd[:5] == ["uv", "tool", "run", "copier", "copy"]
assert ["-d", "database_dialect=postgresql"] == cmd[-2:]
assert cwd == tmp_path
assert extra_env is None
def test_template_check_uses_uv_tool_run_copier(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
calls = []
ctx = ProjectContext(tmp_path, "business", {}, False)
def fake_run(cmd, cwd, extra_env=None):
calls.append((cmd, cwd, extra_env))
return 0
monkeypatch.setattr(cli_module, "require_project", lambda allowed=None: ctx)
monkeypatch.setattr(cli_module, "run", fake_run)
args = build_parser().parse_args(["template", "check"])
assert args.handler(args) == 0
cmd, cwd, extra_env = calls[0]
assert cmd == ["uv", "tool", "run", "copier", "update", "--defaults", "--pretend", "--vcs-ref", "HEAD", str(tmp_path)]
assert cwd == tmp_path
assert extra_env is None
def test_ensure_env_file_copies_example(tmp_path: Path) -> None:
write(tmp_path / ".env.example", "APP_ENV=prod\n")
ensure_env_file(tmp_path)
assert (tmp_path / ".env").read_text(encoding="utf-8") == "APP_ENV=prod\n"
def test_docker_creates_env_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
calls = []
ctx = ProjectContext(tmp_path, "business", {}, False)
write(tmp_path / ".env.example", "APP_ENV=prod\n")
def fake_run(cmd, cwd, extra_env=None):
calls.append((cmd, cwd, extra_env))
return 0
monkeypatch.setattr(cli_module, "require_project", lambda allowed=None: ctx)
monkeypatch.setattr(cli_module, "run", fake_run)
args = build_parser().parse_args(["docker", "up"])
assert args.handler(args) == 0
assert (tmp_path / ".env").read_text(encoding="utf-8") == "APP_ENV=prod\n"
assert calls[0][0] == ["docker", "compose", "up", "-d", "--build"]
def test_project_env_merges_pythonpath(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv("PYTHONPATH", "existing")
env = project_env(tmp_path, {"PYTHONPATH": str(tmp_path)})
assert env["PYTHONPATH"] == os.pathsep.join([str(tmp_path), "existing"])
assert merge_path_env(str(tmp_path), "existing") == env["PYTHONPATH"]
def test_resolve_executable_uses_shutil_which(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(cli_module.shutil, "which", lambda command: r"C:\Tools\uv.exe" if command == "uv" else None)
assert resolve_executable("uv") == r"C:\Tools\uv.exe"
with pytest.raises(CliError, match="command not found: git"):
resolve_executable("git")
def test_run_resolves_executable_before_subprocess(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
calls = []
def fake_resolve(command: str) -> str:
return f"/resolved/{command}"
def fake_subprocess_run(cmd, cwd, env, check):
calls.append((cmd, cwd, env, check))
class Completed:
returncode = 0
return Completed()
monkeypatch.setattr(cli_module, "resolve_executable", fake_resolve)
monkeypatch.setattr(cli_module.subprocess, "run", fake_subprocess_run)
assert cli_module.run(["uv", "--version"], tmp_path) == 0
assert calls[0][0] == ["/resolved/uv", "--version"]
assert calls[0][1] == str(tmp_path)
def test_version_helpers() -> None: def test_version_helpers() -> None:
assert normalize_version("v1.2.3") == "1.2.3" assert normalize_version("v1.2.3") == "1.2.3"
assert version_is_newer("1.2.3", "1.2.4") is True assert version_is_newer("1.2.3", "1.2.4") is True

Loading…
Cancel
Save