feat: add CLI entry point with start/stop/status/config subcommands

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-28 17:52:43 +08:00
parent 81957ca188
commit 5fcb4596e4
2 changed files with 217 additions and 0 deletions

169
src/jianda_proxy/cli.py Normal file
View File

@@ -0,0 +1,169 @@
import argparse
import logging
import os
import sys
from jianda_proxy.config import (
load_config,
set_config_value,
get_config_value,
list_config,
get_config_path,
get_log_path,
)
from jianda_proxy.proxy import create_server
from jianda_proxy.daemon import (
write_pid,
is_running,
stop_daemon,
get_status,
daemonize,
remove_pid,
)
def build_parser():
parser = argparse.ArgumentParser(prog="jianda-proxy", description="HTTP proxy for Gitea LFS")
sub = parser.add_subparsers(dest="command")
# start
p_start = sub.add_parser("start", help="Start the proxy daemon")
p_start.add_argument("--foreground", action="store_true", help="Run in foreground (no daemon)")
p_start.add_argument("--port", type=str, default=None, help="Override listen port")
# stop
sub.add_parser("stop", help="Stop the proxy daemon")
# status
sub.add_parser("status", help="Show proxy daemon status")
# config
p_config = sub.add_parser("config", help="Manage configuration")
config_sub = p_config.add_subparsers(dest="config_command")
p_set = config_sub.add_parser("set", help="Set a config value")
p_set.add_argument("key", help="Config key")
p_set.add_argument("value", help="Config value")
p_get = config_sub.add_parser("get", help="Get a config value")
p_get.add_argument("key", help="Config key")
config_sub.add_parser("list", help="List all config values")
config_sub.add_parser("path", help="Show config file path")
return parser
def cmd_start(args):
if is_running():
print("Error: jianda-proxy is already running.", file=sys.stderr)
sys.exit(1)
config = load_config()
config_dict = dict(config["proxy"])
if args.port:
config_dict["listen_port"] = args.port
if args.foreground:
_run_foreground(config_dict)
else:
_run_daemonized(config_dict)
def _run_foreground(config_dict):
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(name)s] %(levelname)s - %(message)s",
)
server = create_server(config_dict)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
def _run_daemonized(config_dict):
from logging.handlers import RotatingFileHandler
log_path = get_log_path()
handler = RotatingFileHandler(log_path, maxBytes=5 * 1024 * 1024, backupCount=3)
handler.setFormatter(
logging.Formatter("%(asctime)s [%(name)s] %(levelname)s - %(message)s")
)
root = logging.getLogger()
root.setLevel(logging.INFO)
root.addHandler(handler)
daemonize()
pid = os.getpid()
write_pid(pid)
server = create_server(config_dict)
try:
server.serve_forever()
finally:
server.server_close()
remove_pid()
def cmd_stop(_args):
result = stop_daemon()
if result == "not_running":
print("jianda-proxy is not running.", file=sys.stderr)
sys.exit(1)
else:
print("jianda-proxy stopped.")
def cmd_status(_args):
status = get_status()
if status["running"]:
print(f"jianda-proxy is running (PID: {status['pid']})")
else:
print("jianda-proxy is not running.")
def cmd_config(args):
if args.config_command == "set":
set_config_value(args.key, args.value)
print(f"{args.key} = {args.value}")
elif args.config_command == "get":
print(get_config_value(args.key))
elif args.config_command == "list":
for k, v in list_config().items():
print(f"{k} = {v}")
elif args.config_command == "path":
print(get_config_path())
else:
print("Unknown config command. Use set/get/list/path.", file=sys.stderr)
sys.exit(1)
def main():
parser = build_parser()
args = parser.parse_args()
if not args.command:
parser.print_help()
sys.exit(1)
commands = {
"start": cmd_start,
"stop": cmd_stop,
"status": cmd_status,
"config": cmd_config,
}
fn = commands.get(args.command)
if fn:
fn(args)
else:
parser.print_help()
sys.exit(1)
if __name__ == "__main__":
main()

48
tests/test_cli.py Normal file
View File

@@ -0,0 +1,48 @@
import pytest
from jianda_proxy.cli import build_parser
class TestParser:
def test_start_subcommand_exists(self):
parser = build_parser()
args = parser.parse_args(["start"])
assert args.command == "start"
assert args.foreground is False
def test_start_foreground_flag(self):
parser = build_parser()
args = parser.parse_args(["start", "--foreground"])
assert args.foreground is True
def test_stop_subcommand(self):
parser = build_parser()
args = parser.parse_args(["stop"])
assert args.command == "stop"
def test_status_subcommand(self):
parser = build_parser()
args = parser.parse_args(["status"])
assert args.command == "status"
def test_config_set(self):
parser = build_parser()
args = parser.parse_args(["config", "set", "listen_port", "8080"])
assert args.config_command == "set"
assert args.key == "listen_port"
assert args.value == "8080"
def test_config_get(self):
parser = build_parser()
args = parser.parse_args(["config", "get", "listen_port"])
assert args.config_command == "get"
assert args.key == "listen_port"
def test_config_list(self):
parser = build_parser()
args = parser.parse_args(["config", "list"])
assert args.config_command == "list"
def test_config_path(self):
parser = build_parser()
args = parser.parse_args(["config", "path"])
assert args.config_command == "path"