feat: add redmine-cli containerized CLI tool
Merge the redmine-cli work from the EnableAI repo into KNEL-AIMiddleware as the canonical home for AI middleware. This is a standalone CLI (python-redmine 2.5.0) that lists, shows, creates, updates, and closes Redmine issues via docker run. Complements the existing mcp-redmine MCP server (protocol-native) with a direct-invocation tool for interactive use. Verified end-to-end against projects.knownelement.com: whoami, projects, statuses, list, show, create, update, and close all tested successfully. Also records host-cleanliness conventions (docker/tea only on host; custom images to Gitea registry) in AGENTS.md, and adds a CLI Tools section to STATUS.md.
This commit is contained in:
@@ -0,0 +1,389 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Redmine CLI - a thin wrapper around the Redmine REST API.
|
||||
|
||||
Connection details come from the environment:
|
||||
REDMINE_URL base URL of the Redmine instance (e.g. https://redmine.example.com)
|
||||
REDMINE_API_KEY API key of an authenticated user
|
||||
|
||||
Designed to be run inside the redmine-cli Docker container, but works anywhere
|
||||
these environment variables are set.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
|
||||
from redminelib import Redmine
|
||||
from redminelib.exceptions import (
|
||||
AuthError,
|
||||
ResourceNotFoundError,
|
||||
ServerError,
|
||||
ValidationError,
|
||||
ConflictError,
|
||||
NoFileError,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Helpers
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _client():
|
||||
"""Build and return an authenticated Redmine client, or exit with help."""
|
||||
url = os.environ.get("REDMINE_URL", "").strip().rstrip("/")
|
||||
key = os.environ.get("REDMINE_API_KEY", "").strip()
|
||||
missing = [n for n, v in (("REDMINE_URL", url), ("REDMINE_API_KEY", key)) if not v]
|
||||
if missing:
|
||||
sys.stderr.write(
|
||||
"ERROR: missing required environment variable(s): "
|
||||
+ ", ".join(missing)
|
||||
+ "\n"
|
||||
)
|
||||
sys.exit(2)
|
||||
return Redmine(url, key=key)
|
||||
|
||||
|
||||
def _current_user(rm):
|
||||
"""Return the Redmine user behind the current API key."""
|
||||
return rm.user.get("current")
|
||||
|
||||
|
||||
def _err(msg, code=1):
|
||||
sys.stderr.write(f"ERROR: {msg}\n")
|
||||
sys.exit(code)
|
||||
|
||||
|
||||
def _kv(label, value):
|
||||
"""Format a label/value line, omitting falsy values gracefully."""
|
||||
if value in (None, "", []):
|
||||
return None
|
||||
return f"{label:>14}: {value}"
|
||||
|
||||
|
||||
def _name(obj):
|
||||
"""Best-effort human name for a user object."""
|
||||
if obj is None:
|
||||
return "(unassigned)"
|
||||
# python-redmine user objects expose these attributes
|
||||
fname = getattr(obj, "firstname", "") or ""
|
||||
lname = getattr(obj, "lastname", "") or ""
|
||||
full = f"{fname} {lname}".strip()
|
||||
login = getattr(obj, "login", "") or ""
|
||||
mail = getattr(obj, "mail", "") or ""
|
||||
if full:
|
||||
return f"{full} ({login})" if login else full
|
||||
return login or mail or str(obj)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Commands
|
||||
# --------------------------------------------------------------------------- #
|
||||
def cmd_whoami(args):
|
||||
rm = _client()
|
||||
try:
|
||||
u = _current_user(rm)
|
||||
except AuthError:
|
||||
_err("authentication failed - check REDMINE_API_KEY", 2)
|
||||
print(f"id: {u.id}")
|
||||
print(f"login: {getattr(u, 'login', '?')}")
|
||||
print(f"name: {_name(u)}")
|
||||
print(f"mail: {getattr(u, 'mail', '?')}")
|
||||
print(f"admin: {getattr(u, 'admin', False)}")
|
||||
print(f"redmine: {os.environ['REDMINE_URL']}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_projects(args):
|
||||
rm = _client()
|
||||
projects = rm.project.all()
|
||||
print(f"{'ID':<6} {'IDENTIFIER':<24} {'NAME'}")
|
||||
print("-" * 60)
|
||||
for p in projects:
|
||||
print(f"{p.id:<6} {str(p.identifier):<24} {p.name}")
|
||||
print(f"\n{len(projects)} project(s)")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_statuses(args):
|
||||
rm = _client()
|
||||
statuses = rm.issue_status.all()
|
||||
print(f"{'ID':<6} {'NAME':<24} IS_CLOSED")
|
||||
print("-" * 40)
|
||||
for s in statuses:
|
||||
print(f"{s.id:<6} {str(s.name):<24} {bool(s.is_closed)}")
|
||||
print(f"\n{len(statuses)} status(es)")
|
||||
return 0
|
||||
|
||||
|
||||
def _resolve_status_id(rm, status_ref):
|
||||
"""Resolve a status name (case-insensitive) or id to a numeric id."""
|
||||
if status_ref is None:
|
||||
return None
|
||||
if status_ref.isdigit():
|
||||
return int(status_ref)
|
||||
needle = status_ref.strip().lower()
|
||||
for s in rm.issue_status.all():
|
||||
if str(s.name).lower() == needle:
|
||||
return s.id
|
||||
_err(f"unknown status '{status_ref}'. Run 'statuses' to list valid names.")
|
||||
|
||||
|
||||
def _resolve_project_id(rm, project_ref):
|
||||
"""Resolve a project id or identifier to a numeric id."""
|
||||
if project_ref is None:
|
||||
return None
|
||||
if project_ref.isdigit():
|
||||
return int(project_ref)
|
||||
needle = project_ref.strip().lower()
|
||||
for p in rm.project.all():
|
||||
if str(p.identifier).lower() == needle:
|
||||
return p.id
|
||||
_err(f"unknown project '{project_ref}'. Run 'projects' to list identifiers.")
|
||||
|
||||
|
||||
def cmd_list(args):
|
||||
rm = _client()
|
||||
filters = {}
|
||||
if args.assigned_to_me:
|
||||
me = _current_user(rm)
|
||||
filters["assigned_to_id"] = me.id
|
||||
elif args.assigned_to is not None:
|
||||
filters["assigned_to_id"] = args.assigned_to
|
||||
if args.project is not None:
|
||||
filters["project_id"] = _resolve_project_id(rm, args.project)
|
||||
if args.status is not None:
|
||||
# status can be a name or id
|
||||
if args.status.lower() in ("open", "open*"):
|
||||
filters["status_id"] = "open"
|
||||
elif args.status.lower() in ("closed", "closed*"):
|
||||
filters["status_id"] = "closed"
|
||||
else:
|
||||
filters["status_id"] = _resolve_status_id(rm, args.status)
|
||||
filters["limit"] = args.limit
|
||||
filters["sort"] = args.sort
|
||||
|
||||
issues = rm.issue.filter(**filters)
|
||||
print(f"{'ID':<8} {'STATUS':<14} {'PRJ':<14} {'DONE':>5} SUBJECT")
|
||||
print("-" * 90)
|
||||
for i in issues:
|
||||
proj = getattr(i, "project", None)
|
||||
proj_id = str(getattr(proj, "identifier", getattr(proj, "id", ""))) if proj else ""
|
||||
status = getattr(i, "status", None)
|
||||
status_name = str(getattr(status, "name", "")) if status else ""
|
||||
done = getattr(i, "done_ratio", 0) or 0
|
||||
print(f"{i.id:<8} {status_name:<14} {proj_id:<14} {done:>4}% {i.subject}")
|
||||
print(f"\n{len(issues)} issue(s)")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_show(args):
|
||||
rm = _client()
|
||||
try:
|
||||
i = rm.issue.get(args.issue_id)
|
||||
except ResourceNotFoundError:
|
||||
_err(f"issue #{args.issue_id} not found")
|
||||
lines = []
|
||||
lines.append(f"#{i.id}: {i.subject}")
|
||||
lines.append("=" * 90)
|
||||
proj = getattr(i, "project", None)
|
||||
status = getattr(i, "status", None)
|
||||
tracker = getattr(i, "tracker", None)
|
||||
priority = getattr(i, "priority", None)
|
||||
lines.append(_kv("project", str(getattr(proj, "name", "")) if proj else None))
|
||||
lines.append(_kv("tracker", str(getattr(tracker, "name", "")) if tracker else None))
|
||||
lines.append(_kv("status", str(getattr(status, "name", "")) if status else None))
|
||||
lines.append(_kv("priority", str(getattr(priority, "name", "")) if priority else None))
|
||||
lines.append(_kv("author", _name(getattr(i, "author", None))))
|
||||
lines.append(_kv("assigned", _name(getattr(i, "assigned_to", None))))
|
||||
lines.append(_kv("done", f"{getattr(i, 'done_ratio', 0) or 0}%"))
|
||||
lines.append(_kv("created", getattr(i, "created_on", None)))
|
||||
lines.append(_kv("updated", getattr(i, "updated_on", None)))
|
||||
lines.append(_kv("start", getattr(i, "start_date", None)))
|
||||
lines.append(_kv("due", getattr(i, "due_date", None)))
|
||||
lines.append(_kv("estimated", getattr(i, "estimated_hours", None)))
|
||||
lines.append(_kv("spent", getattr(i, "spent_hours", None)))
|
||||
for ln in (l for l in lines if l):
|
||||
print(ln)
|
||||
|
||||
desc = getattr(i, "description", "") or ""
|
||||
if desc.strip():
|
||||
print("\n--- description ---")
|
||||
print(desc.strip())
|
||||
|
||||
if not args.no_journals:
|
||||
journals = getattr(i, "journals", []) or []
|
||||
notes = [j for j in journals if getattr(j, "notes", "") and str(j.notes).strip()]
|
||||
if notes:
|
||||
print("\n--- history (notes) ---")
|
||||
for j in notes:
|
||||
who = _name(getattr(j, "user", None))
|
||||
when = getattr(j, "created_on", "?")
|
||||
print(f"\n[{when}] {who}:")
|
||||
print(str(j.notes).strip())
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_create(args):
|
||||
rm = _client()
|
||||
if not args.project:
|
||||
_err("--project is required to create an issue")
|
||||
if not args.subject:
|
||||
_err("--subject is required to create an issue")
|
||||
fields = {
|
||||
"project_id": _resolve_project_id(rm, args.project),
|
||||
"subject": args.subject,
|
||||
}
|
||||
if args.description:
|
||||
fields["description"] = args.description
|
||||
if args.assigned_to is not None:
|
||||
fields["assigned_to_id"] = args.assigned_to
|
||||
if args.tracker is not None:
|
||||
fields["tracker_id"] = args.tracker
|
||||
if args.priority is not None:
|
||||
fields["priority_id"] = args.priority
|
||||
if args.status is not None:
|
||||
sid = _resolve_status_id(rm, args.status)
|
||||
if sid:
|
||||
fields["status_id"] = sid
|
||||
try:
|
||||
issue = rm.issue.create(**fields)
|
||||
except ValidationError as e:
|
||||
_err(f"validation failed: {e}")
|
||||
print(f"Created issue #{issue.id}: {issue.subject}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_update(args):
|
||||
rm = _client()
|
||||
fields = {}
|
||||
if args.status is not None:
|
||||
sid = _resolve_status_id(rm, args.status)
|
||||
if sid:
|
||||
fields["status_id"] = sid
|
||||
if args.notes:
|
||||
fields["notes"] = args.notes
|
||||
if args.assigned_to is not None:
|
||||
fields["assigned_to_id"] = args.assigned_to
|
||||
if args.done_ratio is not None:
|
||||
fields["done_ratio"] = args.done_ratio
|
||||
if args.subject is not None:
|
||||
fields["subject"] = args.subject
|
||||
if args.priority is not None:
|
||||
fields["priority_id"] = args.priority
|
||||
if not fields:
|
||||
_err("no fields to update; pass at least one of --status/--notes/--assigned-to/--done-ratio/--subject/--priority")
|
||||
try:
|
||||
rm.issue.update(args.issue_id, **fields)
|
||||
except ResourceNotFoundError:
|
||||
_err(f"issue #{args.issue_id} not found")
|
||||
except ValidationError as e:
|
||||
_err(f"validation failed: {e}")
|
||||
print(f"Updated issue #{args.issue_id}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_close(args):
|
||||
rm = _client()
|
||||
close_id = None
|
||||
for s in rm.issue_status.all():
|
||||
if getattr(s, "is_closed", False):
|
||||
close_id = s.id
|
||||
break
|
||||
if close_id is None:
|
||||
_err("no closed status found on this Redmine instance")
|
||||
fields = {"status_id": close_id}
|
||||
if args.notes:
|
||||
fields["notes"] = args.notes
|
||||
if args.done_ratio is not None:
|
||||
fields["done_ratio"] = args.done_ratio
|
||||
else:
|
||||
fields["done_ratio"] = 100
|
||||
try:
|
||||
rm.issue.update(args.issue_id, **fields)
|
||||
except ResourceNotFoundError:
|
||||
_err(f"issue #{args.issue_id} not found")
|
||||
except ValidationError as e:
|
||||
_err(f"validation failed: {e}")
|
||||
print(f"Closed issue #{args.issue_id} (status_id={close_id}, done=100%)")
|
||||
return 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Argument parsing
|
||||
# --------------------------------------------------------------------------- #
|
||||
def build_parser():
|
||||
p = argparse.ArgumentParser(
|
||||
prog="redmine-cli",
|
||||
description="Access, edit, and close Redmine issues via the REST API.",
|
||||
)
|
||||
sub = p.add_subparsers(dest="command", required=True)
|
||||
|
||||
sub.add_parser("whoami", help="show the authenticated user (connection test)").set_defaults(func=cmd_whoami)
|
||||
|
||||
sp = sub.add_parser("projects", help="list projects")
|
||||
sp.set_defaults(func=cmd_projects)
|
||||
|
||||
sp = sub.add_parser("statuses", help="list issue statuses")
|
||||
sp.set_defaults(func=cmd_statuses)
|
||||
|
||||
sp = sub.add_parser("list", aliases=["ls"], help="list issues")
|
||||
sp.add_argument("-m", "--assigned-to-me", action="store_true", help="only issues assigned to the current user")
|
||||
sp.add_argument("-a", "--assigned-to", metavar="USER_ID", help="filter by assignee user id")
|
||||
sp.add_argument("-p", "--project", metavar="ID_OR_IDENTIFIER", help="filter by project")
|
||||
sp.add_argument("-s", "--status", metavar="NAME_OR_ID", help="filter by status (name/id, or 'open'/'closed')")
|
||||
sp.add_argument("-l", "--limit", type=int, default=50, help="max issues to return (default 50)")
|
||||
sp.add_argument("--sort", default="priority:desc,updated_on:desc", help="sort order")
|
||||
sp.set_defaults(func=cmd_list)
|
||||
|
||||
sp = sub.add_parser("show", help="show details of an issue")
|
||||
sp.add_argument("issue_id", type=int)
|
||||
sp.add_argument("--no-journals", action="store_true", help="omit note history")
|
||||
sp.set_defaults(func=cmd_show)
|
||||
|
||||
sp = sub.add_parser("create", help="create a new issue")
|
||||
sp.add_argument("-p", "--project", required=True, metavar="ID_OR_IDENTIFIER")
|
||||
sp.add_argument("-s", "--subject", required=True)
|
||||
sp.add_argument("-d", "--description")
|
||||
sp.add_argument("-a", "--assigned-to", metavar="USER_ID")
|
||||
sp.add_argument("-t", "--tracker", metavar="TRACKER_ID")
|
||||
sp.add_argument("--priority", metavar="PRIORITY_ID")
|
||||
sp.add_argument("--status", metavar="NAME_OR_ID")
|
||||
sp.set_defaults(func=cmd_create)
|
||||
|
||||
sp = sub.add_parser("update", help="update an issue")
|
||||
sp.add_argument("issue_id", type=int)
|
||||
sp.add_argument("-s", "--status", metavar="NAME_OR_ID", help="new status")
|
||||
sp.add_argument("-n", "--notes", help="add a note/journal comment")
|
||||
sp.add_argument("-a", "--assigned-to", metavar="USER_ID")
|
||||
sp.add_argument("--done-ratio", type=int, metavar="0-100")
|
||||
sp.add_argument("--subject")
|
||||
sp.add_argument("--priority", metavar="PRIORITY_ID")
|
||||
sp.set_defaults(func=cmd_update)
|
||||
|
||||
sp = sub.add_parser("close", help="close an issue (set to first closed status, done ratio 100%%)")
|
||||
sp.add_argument("issue_id", type=int)
|
||||
sp.add_argument("-n", "--notes", help="add a note/journal comment")
|
||||
sp.add_argument("--done-ratio", type=int, metavar="0-100", help="override done ratio (default 100)")
|
||||
sp.set_defaults(func=cmd_close)
|
||||
|
||||
return p
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
parser = build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
try:
|
||||
return args.func(args)
|
||||
except AuthError:
|
||||
_err("authentication failed - check REDMINE_API_KEY", 2)
|
||||
except ServerError as e:
|
||||
_err(f"server error: {e}")
|
||||
except ConflictError as e:
|
||||
_err(f"conflict: {e}")
|
||||
except (NoFileError, ResourceNotFoundError) as e:
|
||||
_err(str(e))
|
||||
except Exception as e: # noqa: BLE001 - top-level safety net for the CLI
|
||||
_err(f"{type(e).__name__}: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user