forked from username/gitea-ci
add set command for repository visibility
This commit is contained in:
@@ -223,6 +223,66 @@ class GiteaAPI:
|
||||
print(f"{Color.RED}Connection error: {e.reason}{Color.RST}")
|
||||
return False
|
||||
|
||||
def _patch(self, endpoint: str, data: dict, retries: int = 3) -> dict | None:
|
||||
"""Make PATCH request with JSON body."""
|
||||
url = f"{self.base_url}{endpoint}"
|
||||
headers = {
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
if self.config.token:
|
||||
headers["Authorization"] = f"token {self.config.token}"
|
||||
else:
|
||||
print(f"{Color.RED}Authentication required for this operation{Color.RST}")
|
||||
return None
|
||||
|
||||
body = json.dumps(data).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=body, headers=headers, method="PATCH")
|
||||
|
||||
for attempt in range(retries):
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
self._retry_delay = 1
|
||||
if resp.status == 204:
|
||||
return {}
|
||||
content = resp.read().decode()
|
||||
return json.loads(content) if content else {}
|
||||
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 401:
|
||||
print(f"{Color.RED}Authentication failed. Check your token.{Color.RST}")
|
||||
return None
|
||||
elif e.code == 403:
|
||||
print(f"{Color.RED}Permission denied. Token may lack required scopes.{Color.RST}")
|
||||
return None
|
||||
elif e.code == 404:
|
||||
print(f"{Color.RED}Not found: {endpoint}{Color.RST}")
|
||||
return None
|
||||
elif e.code == 422:
|
||||
print(f"{Color.RED}Invalid request{Color.RST}")
|
||||
return None
|
||||
elif e.code == 429:
|
||||
if attempt < retries - 1:
|
||||
delay = self._retry_delay
|
||||
print(f"{Color.YLW}Rate limited. Waiting {delay}s...{Color.RST}")
|
||||
time.sleep(delay)
|
||||
self._retry_delay = min(self._retry_delay * 2, self._max_retry_delay)
|
||||
continue
|
||||
return None
|
||||
else:
|
||||
print(f"{Color.RED}API error: {e.code} {e.reason}{Color.RST}")
|
||||
return None
|
||||
|
||||
except urllib.error.URLError as e:
|
||||
if attempt < retries - 1:
|
||||
time.sleep(self._retry_delay)
|
||||
continue
|
||||
print(f"{Color.RED}Connection error: {e.reason}{Color.RST}")
|
||||
return None
|
||||
|
||||
return None
|
||||
|
||||
def throttle(self) -> None:
|
||||
"""Add small delay between requests to avoid rate limiting."""
|
||||
if self._rate_limited:
|
||||
@@ -255,6 +315,10 @@ class GiteaAPI:
|
||||
"""Check if a repository exists."""
|
||||
return self.get_repo(owner, repo) is not None
|
||||
|
||||
def set_repo_visibility(self, owner: str, repo: str, private: bool) -> dict | None:
|
||||
"""Set repository visibility (private/public)."""
|
||||
return self._patch(f"/repos/{owner}/{repo}", {"private": private})
|
||||
|
||||
# Workflow run methods
|
||||
def get_runs(self, owner: str, repo: str, limit: int = 10) -> list:
|
||||
"""Get workflow runs for a repository."""
|
||||
|
||||
@@ -12,7 +12,7 @@ from .commands import (
|
||||
cmd_trigger, cmd_rerun, cmd_cancel, cmd_validate, cmd_workflows,
|
||||
cmd_artifacts,
|
||||
cmd_runners, cmd_register_token,
|
||||
cmd_stats, cmd_pr, cmd_compare, cmd_infra, cmd_config, cmd_repo,
|
||||
cmd_stats, cmd_pr, cmd_compare, cmd_infra, cmd_config, cmd_repo, cmd_set,
|
||||
)
|
||||
|
||||
|
||||
@@ -69,6 +69,7 @@ Commands:
|
||||
workflows List workflows in a repository
|
||||
infra Show infrastructure status
|
||||
config Configure token and defaults
|
||||
set Set repository visibility (private/public)
|
||||
|
||||
Environment:
|
||||
GITEA_URL Gitea instance URL
|
||||
@@ -109,6 +110,8 @@ Examples:
|
||||
gitea-ci validate Validate local workflow files
|
||||
gitea-ci validate --check-runners Cross-reference runs-on labels
|
||||
gitea-ci validate --strict Treat warnings as errors
|
||||
gitea-ci set user/repo private Make repository private
|
||||
gitea-ci set user/repo public Make repository public
|
||||
""",
|
||||
)
|
||||
|
||||
@@ -185,6 +188,11 @@ Examples:
|
||||
repo_p.add_argument("-p", "--private", action="store_true", help="Make repository private")
|
||||
add_repo_args(repo_p)
|
||||
|
||||
# set
|
||||
set_p = subparsers.add_parser("set", help="Set repository visibility")
|
||||
set_p.add_argument("repo_path", help="Repository as owner/repo")
|
||||
set_p.add_argument("value", choices=["private", "public"], help="Visibility: private or public")
|
||||
|
||||
# trigger
|
||||
trigger_p = subparsers.add_parser("trigger", aliases=["t"], help="Trigger workflow")
|
||||
add_repo_args(trigger_p)
|
||||
@@ -317,6 +325,13 @@ Examples:
|
||||
return cmd_config(api, args)
|
||||
elif args.command == "repo":
|
||||
return cmd_repo(api, args)
|
||||
elif args.command == "set":
|
||||
# Parse repo_path for set command
|
||||
if hasattr(args, "repo_path") and args.repo_path and "/" in args.repo_path:
|
||||
parts = args.repo_path.split("/", 1)
|
||||
args.owner = parts[0]
|
||||
args.repo = parts[1]
|
||||
return cmd_set(api, args)
|
||||
elif args.command in ("trigger", "t"):
|
||||
return cmd_trigger(api, args)
|
||||
elif args.command == "rerun":
|
||||
|
||||
@@ -4,7 +4,7 @@ from .runs import cmd_list, cmd_status, cmd_logs, cmd_watch, cmd_delete, cmd_dis
|
||||
from .workflows import cmd_trigger, cmd_rerun, cmd_cancel, cmd_validate, cmd_workflows
|
||||
from .artifacts import cmd_artifacts
|
||||
from .runners import cmd_runners, cmd_register_token
|
||||
from .inspect import cmd_stats, cmd_pr, cmd_compare, cmd_infra, cmd_config, cmd_repo
|
||||
from .inspect import cmd_stats, cmd_pr, cmd_compare, cmd_infra, cmd_config, cmd_repo, cmd_set
|
||||
|
||||
__all__ = [
|
||||
# runs
|
||||
@@ -32,4 +32,5 @@ __all__ = [
|
||||
"cmd_infra",
|
||||
"cmd_config",
|
||||
"cmd_repo",
|
||||
"cmd_set",
|
||||
]
|
||||
|
||||
@@ -840,3 +840,40 @@ def cmd_repo(api: GiteaAPI, args: argparse.Namespace) -> int:
|
||||
return 0
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_set(api: GiteaAPI, args: argparse.Namespace) -> int:
|
||||
"""Set repository properties (visibility)."""
|
||||
owner = args.owner
|
||||
repo = args.repo
|
||||
value = getattr(args, "value", None)
|
||||
|
||||
if not owner or not repo:
|
||||
print(f"{Color.RED}Repository (owner/repo) required{Color.RST}")
|
||||
return 1
|
||||
|
||||
if value not in ("private", "public"):
|
||||
print(f"{Color.RED}Value must be 'private' or 'public'{Color.RST}")
|
||||
return 1
|
||||
|
||||
# Check current state
|
||||
repo_info = api.get_repo(owner, repo)
|
||||
if not repo_info:
|
||||
print(f"{Color.RED}Repository {owner}/{repo} not found{Color.RST}")
|
||||
return 1
|
||||
|
||||
current = "private" if repo_info.get("private") else "public"
|
||||
if current == value:
|
||||
print(f"{Color.DIM}✓ {owner}/{repo} is already {value}{Color.RST}")
|
||||
return 0
|
||||
|
||||
# Update visibility
|
||||
private = value == "private"
|
||||
result = api.set_repo_visibility(owner, repo, private)
|
||||
if result is not None:
|
||||
icon = Color.YLW if private else Color.GRN
|
||||
print(f"{Color.GRN}✓{Color.RST} {owner}/{repo} is now {icon}{value}{Color.RST}")
|
||||
return 0
|
||||
else:
|
||||
print(f"{Color.RED}✗{Color.RST} Failed to update visibility")
|
||||
return 1
|
||||
|
||||
Reference in New Issue
Block a user