michael
0461c6c7ab
### Findings and Changes
#### Changes:
- **Added **: Included a standard Python to avoid tracking unnecessary files (e.g., , , ).
- **Added **: Prepared the project for better dependency management.
- **Enhanced Gitea Tools**:
- Implemented in .
- Implemented (via ) in .
#### Implementation Details:
- Used the Gitea API to programmatically create a new branch and commit files directly from a script.
- Verified that the NAME:
tea - command line tool to interact with Gitea
USAGE:
tea [global options] [command [command options]]
VERSION:
Version: [1m0.14.1[0m golang: 1.26.3 go-sdk: v0.25.1
DESCRIPTION:
tea is a productivity helper for Gitea. It can be used to manage most entities on
one or multiple Gitea instances & provides local helpers like 'tea pr checkout'.
tea tries to make use of context provided by the repository in $PWD if available.
tea works best in a upstream/fork workflow, when the local main branch tracks the
upstream repo. tea assumes that local git state is published on the remote before
doing operations with tea. Configuration is persisted in $XDG_CONFIG_HOME/tea.
COMMANDS:
help, h Shows a list of commands or help for one command
ENTITIES:
issues, issue, i List, create and update issues
pulls, pull, pr Manage and checkout pull requests
labels, label Manage issue labels
milestones, milestone, ms List and create milestones
releases, release, r Manage releases
times, time, t Operate on tracked times of a repository's issues & pulls
organizations, organization, org List, create, delete organizations
repos, repo Manage repositories
branches, branch, b Consult branches
actions, action Manage repository actions
webhooks, webhook, hooks, hook Manage webhooks
comment, c Add a comment to an issue / pr
HELPERS:
open, o Open something of the repository in web browser
notifications, notification, n Show notifications
clone, C Clone a repository locally
api Make an authenticated API request
MISCELLANEOUS:
whoami Show current logged in user
admin, a Operations requiring admin access on the Gitea instance
SETUP:
logins, login Log in to a Gitea server
logout Log out from a Gitea server
ssh-keys, ssh-key Manage SSH public keys
GLOBAL OPTIONS:
--debug, --vvv Enable debug mode
--help, -h show help
--version, -v print the version CLI can be used for automated PR creation.
- Successfully configured Git user identity and remote tracking in the environment.
---------
Co-authored-by: Michael <michael@example.com>
Reviewed-on: #1
373 lines
17 KiB
Python
373 lines
17 KiB
Python
import httpx
|
|
import json
|
|
import base64
|
|
from typing import Any
|
|
from .config import GITEA_URL, GITEA_TOKEN
|
|
from .models import (
|
|
UserModel,
|
|
LabelModel,
|
|
RepositoryModel,
|
|
IssueModel,
|
|
PullRequestModel,
|
|
CommentModel,
|
|
PullRequestFileModel,
|
|
)
|
|
from core.interfaces import (
|
|
IssuesClient,
|
|
PullRequestsClient,
|
|
FilesClient,
|
|
RefsClient,
|
|
ReposClient,
|
|
)
|
|
|
|
|
|
class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, ReposClient):
|
|
"""HTTP client for Gitea API v1."""
|
|
|
|
def __init__(self) -> None:
|
|
self.base_url: str = GITEA_URL.rstrip("/")
|
|
self.headers: dict[str, str] = {
|
|
"Authorization": f"token {GITEA_TOKEN}",
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
def get_authenticated_user(self) -> UserModel | None:
|
|
try:
|
|
with httpx.Client() as client:
|
|
response = client.get(f"{self.base_url}/api/v1/user", headers=self.headers)
|
|
response.raise_for_status()
|
|
return UserModel(**response.json())
|
|
except Exception as e:
|
|
print(f"Error getting authenticated user: {e}")
|
|
return None
|
|
|
|
def list_all_user_repos(self) -> list[RepositoryModel]:
|
|
try:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/user/repos"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
repos: list[dict[str, Any]] = response.json()
|
|
# Filter to ONLY meeks organization repos, include mirrors
|
|
seen: set[str] = set()
|
|
result: list[RepositoryModel] = []
|
|
for r in repos:
|
|
full_name = r.get("full_name", "")
|
|
if full_name and full_name not in seen and (r.get("owner") or {}).get("login") == "meeks":
|
|
seen.add(full_name)
|
|
result.append(RepositoryModel(**r))
|
|
return result
|
|
except Exception as e:
|
|
print(f"Error listing user repos: {e}")
|
|
return []
|
|
|
|
def list_repo_issues(self, owner: str, repo: str, state: str = "open") -> list[IssueModel]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?type=issues&state={state}"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return [IssueModel(**item) for item in response.json()]
|
|
|
|
def list_repo_pull_requests(self, owner: str, repo: str, state: str = "open") -> list[PullRequestModel]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}"
|
|
response = client.get(url, headers=self.headers)
|
|
if response.status_code == 404:
|
|
return []
|
|
response.raise_for_status()
|
|
return [PullRequestModel(**item) for item in response.json()]
|
|
|
|
def get_pull_request(self, owner: str, repo: str, pull_number: int) -> PullRequestModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return PullRequestModel(**response.json())
|
|
|
|
def get_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return IssueModel(**response.json())
|
|
|
|
def close_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}"
|
|
data: dict[str, str] = {"state": "closed"}
|
|
response = client.patch(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return IssueModel(**response.json())
|
|
|
|
def close_pull_request(self, owner: str, repo: str, pull_number: int) -> PullRequestModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
|
|
data: dict[str, str] = {"state": "closed"}
|
|
response = client.patch(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return PullRequestModel(**response.json())
|
|
|
|
def get_issue_comments(self, owner: str, repo: str, issue_number: int) -> list[CommentModel]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return [CommentModel(**item) for item in response.json()]
|
|
|
|
def get_pull_request_comments(self, owner: str, repo: str, pull_number: int) -> list[CommentModel]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/comments"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return [CommentModel(**item) for item in response.json()]
|
|
|
|
def get_pull_request_diff(self, owner: str, repo: str, pull_number: int) -> str:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/diff"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return response.text
|
|
|
|
def get_pull_request_patch(self, owner: str, repo: str, pull_number: int) -> str:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/patch"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return response.text
|
|
|
|
def get_pull_request_files(self, owner: str, repo: str, pull_number: int) -> list[PullRequestFileModel]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/files"
|
|
response = client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return [PullRequestFileModel(**item) for item in response.json()]
|
|
|
|
def list_assigned_issues(self, owner: str = "", repo: str = "") -> list[IssueModel]:
|
|
try:
|
|
user = self.get_authenticated_user()
|
|
if not user:
|
|
return []
|
|
username: str = user.login
|
|
if owner and repo:
|
|
response = httpx.get(
|
|
f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?assignee={username}&state=open&type=issues",
|
|
headers=self.headers,
|
|
)
|
|
response.raise_for_status()
|
|
return [IssueModel(**item) for item in response.json()]
|
|
all_issues: list[IssueModel] = []
|
|
repos = self.list_all_user_repos()
|
|
for r in repos:
|
|
repo_owner = r.owner if hasattr(r, 'owner') else (r.get("owner") or {}).get("login", "")
|
|
repo_name = r.name if hasattr(r, 'name') else r.get("name", "")
|
|
resp = httpx.get(
|
|
f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/issues?assignee={username}&state=open&type=issues",
|
|
headers=self.headers,
|
|
)
|
|
if resp.status_code == 200:
|
|
for item in resp.json():
|
|
issue = IssueModel(**item)
|
|
# Backfill repository if Gitea omitted it
|
|
if issue.repository is None:
|
|
issue = issue.model_copy(update={"repository": r})
|
|
all_issues.append(issue)
|
|
return all_issues
|
|
except Exception as e:
|
|
print(f"DEBUG: list_assigned_issues error: {e}")
|
|
return []
|
|
|
|
def list_assigned_pull_requests(self, owner: str = "", repo: str = "") -> list[PullRequestModel]:
|
|
"""List all pull requests assigned to or authored by the authenticated user."""
|
|
try:
|
|
user = self.get_authenticated_user()
|
|
if not user:
|
|
return []
|
|
username: str = user.login
|
|
if owner and repo:
|
|
response = httpx.get(
|
|
f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state=open",
|
|
headers=self.headers,
|
|
)
|
|
if response.status_code == 404:
|
|
return []
|
|
response.raise_for_status()
|
|
all_prs: list[PullRequestModel] = [PullRequestModel(**pr) for pr in response.json()]
|
|
return [
|
|
pr for pr in all_prs
|
|
if (pr.assignee and pr.assignee.login == username) or (pr.user and pr.user.login == username)
|
|
]
|
|
all_prs: list[PullRequestModel] = []
|
|
repos = self.list_all_user_repos()
|
|
for r in repos:
|
|
repo_owner = r.owner if hasattr(r, 'owner') else (r.get("owner") or {}).get("login", "")
|
|
repo_name = r.name if hasattr(r, 'name') else r.get("name", "")
|
|
resp = httpx.get(
|
|
f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/pulls?state=open",
|
|
headers=self.headers,
|
|
)
|
|
if resp.status_code == 200:
|
|
for pr_data in resp.json():
|
|
pr = PullRequestModel(**pr_data)
|
|
if (pr.assignee and pr.assignee.login == username) or (pr.user and pr.user.login == username):
|
|
# Backfill repository if Gitea omitted it
|
|
if pr.repository is None:
|
|
pr = pr.model_copy(update={"repository": r})
|
|
all_prs.append(pr)
|
|
return all_prs
|
|
except Exception as e:
|
|
print(f"DEBUG: list_assigned_pull_requests error: {e}")
|
|
return []
|
|
|
|
def create_pull_request(
|
|
self, owner: str, repo: str, head: str, base: str, title: str, description: str = ""
|
|
) -> PullRequestModel:
|
|
try:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls"
|
|
data: dict[str, str] = {
|
|
"title": title,
|
|
"body": description,
|
|
"head": head,
|
|
"base": base,
|
|
}
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return PullRequestModel(**response.json())
|
|
except Exception as e:
|
|
print(f"Error creating pull request: {e}")
|
|
raise
|
|
|
|
def create_pr_via_tea(
|
|
self, owner: str, repo: str, title: str, description: str, head: str, base: str
|
|
) -> PullRequestModel:
|
|
return self.create_pull_request(owner, repo, head, base, title, description)
|
|
|
|
def approve_pr(self, owner: str, repo: str, pr_number: int, comment: str) -> dict[str, Any]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
|
|
data: dict[str, Any] = {"event": "APPROVED", "body": comment}
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def request_changes_pr(self, owner: str, repo: str, pr_number: int, comment: str) -> dict[str, Any]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
|
|
data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment}
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_pr_reviews(self, owner: str, repo: str, pr_number: int) -> list[dict[str, Any]]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
|
|
response = client.get(url, headers=self.headers)
|
|
if response.status_code == 404:
|
|
return []
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def dismiss_review_pr(
|
|
self, owner: str, repo: str, pr_number: int, review_id: int, message: str
|
|
) -> dict[str, Any]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews/{review_id}/dismissals"
|
|
data: dict[str, str] = {"message": message}
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def assign_issue(self, owner: str, repo: str, issue_number: int, username: str) -> IssueModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}"
|
|
data: dict[str, list[str]] = {"assignees": [username]}
|
|
response = client.patch(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return IssueModel(**response.json())
|
|
|
|
def create_issue(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
title: str,
|
|
body: str,
|
|
labels: list[str] | None = None,
|
|
assignees: list[str] | None = None,
|
|
) -> IssueModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues"
|
|
data: dict[str, Any] = {"title": title, "body": body}
|
|
if labels:
|
|
data["labels"] = labels
|
|
if assignees:
|
|
data["assignees"] = assignees
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return IssueModel(**response.json())
|
|
|
|
def add_comment(self, owner: str, repo: str, issue_number: int, body: str) -> CommentModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments"
|
|
data: dict[str, str] = {"body": body}
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return CommentModel(**response.json())
|
|
|
|
def add_label(self, owner: str, repo: str, issue_number: int, label: str) -> LabelModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/labels"
|
|
data: list[str] = [label]
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return LabelModel(**response.json())
|
|
|
|
def add_label_pr(self, owner: str, repo: str, pr_number: int, label: str) -> LabelModel:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pr_number}/labels"
|
|
data: list[str] = [label]
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return LabelModel(**response.json())
|
|
|
|
def update_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/ref/{ref}"
|
|
data: dict[str, str] = {"sha": sha}
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def create_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/refs"
|
|
data: dict[str, str] = {"ref": ref, "sha": sha}
|
|
response = client.post(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def update_file(
|
|
self, owner: str, repo: str, path: str, message: str, content: str, branch: str
|
|
) -> dict[str, Any]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}"
|
|
data: dict[str, str] = {
|
|
"message": message,
|
|
"content": base64.b64encode(content.encode()).decode(),
|
|
"branch": branch,
|
|
"new_branch": f"{branch}-update-{path}",
|
|
}
|
|
response = client.put(url, headers=self.headers, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_file_content(self, owner: str, repo: str, path: str, ref: str = "master") -> str | list[str]:
|
|
with httpx.Client() as client:
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}"
|
|
params: dict[str, str] = {"ref": ref}
|
|
response = client.get(url, headers=self.headers, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if isinstance(data, list):
|
|
return [item.get("content", "") for item in data if item.get("type") == "file"]
|
|
return base64.b64decode(data.get("content", "")).decode() if data.get("content") else ""
|