michael/coding-agent-gitea
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix http

Browse Source
This commit is contained in:
Michael Ingvarsson
2026-07-16 12:21:56 +02:00
parent 9e40e7fed8
commit e54b5f1848
2 changed files with 215 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea/client.py
+213 -231
View File
@@ -34,117 +34,121 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
"Authorization": f"token {GITEA_TOKEN}", "Authorization": f"token {GITEA_TOKEN}",
"Accept": "application/json", "Accept": "application/json",
} }
self.client: httpx.Client = httpx.Client(headers=self.headers)
def close(self) -> None:
"""Close the underlying HTTP client."""
self.client.close()
def __enter__(self) -> "GiteaClient":
return self
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
self.close()
def __del__(self) -> None:
try:
self.client.close()
except Exception:
pass
def get_authenticated_user(self) -> UserModel | None: def get_authenticated_user(self) -> UserModel | None:
try: try:
with httpx.Client() as client: response = self.client.get(f"{self.base_url}/api/v1/user")
response = client.get(f"{self.base_url}/api/v1/user", headers=self.headers) response.raise_for_status()
response.raise_for_status() return UserModel(**response.json())
return UserModel(**response.json())
except Exception as e: except Exception as e:
logger.error(f"Error getting authenticated user: {e}", exc_info=True) logger.error(f"Error getting authenticated user: {e}", exc_info=True)
return None return None
def list_all_user_repos(self) -> list[RepositoryModel]: def list_all_user_repos(self) -> list[RepositoryModel]:
try: try:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/user/repos"
url = f"{self.base_url}/api/v1/user/repos" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() repos: list[dict[str, Any]] = response.json()
repos: list[dict[str, Any]] = response.json() # Filter to ONLY meeks organization repos, include mirrors
# Filter to ONLY meeks organization repos, include mirrors seen: set[str] = set()
seen: set[str] = set() result: list[RepositoryModel] = []
result: list[RepositoryModel] = [] for r in repos:
for r in repos: full_name = r.get("full_name", "")
full_name = r.get("full_name", "") if full_name and full_name not in seen and (r.get("owner") or {}).get("login") == "meeks":
if full_name and full_name not in seen and (r.get("owner") or {}).get("login") == "meeks": seen.add(full_name)
seen.add(full_name) result.append(RepositoryModel(**r))
result.append(RepositoryModel(**r)) return result
return result
except Exception as e: except Exception as e:
logger.error(f"Error listing user repos: {e}", exc_info=True) logger.error(f"Error listing user repos: {e}", exc_info=True)
return [] return []
def list_repo_issues(self, owner: str, repo: str, state: str = "open") -> list[IssueModel]: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?type=issues&state={state}" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return [IssueModel(**item) for item in response.json()]
return [IssueModel(**item) for item in response.json()]
def list_repo_pull_requests(self, owner: str, repo: str, state: str = "open") -> list[PullRequestModel]: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}" response = self.client.get(url)
response = client.get(url, headers=self.headers) if response.status_code == 404:
if response.status_code == 404: return []
return [] response.raise_for_status()
response.raise_for_status() return [PullRequestModel(**item) for item in response.json()]
return [PullRequestModel(**item) for item in response.json()]
def get_pull_request(self, owner: str, repo: str, pull_number: int) -> PullRequestModel: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return PullRequestModel(**response.json())
return PullRequestModel(**response.json())
def get_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return IssueModel(**response.json())
return IssueModel(**response.json())
def close_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" data: dict[str, str] = {"state": "closed"}
data: dict[str, str] = {"state": "closed"} response = self.client.patch(url, json=data)
response = client.patch(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return IssueModel(**response.json())
return IssueModel(**response.json())
def close_pull_request(self, owner: str, repo: str, pull_number: int) -> PullRequestModel: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" data: dict[str, str] = {"state": "closed"}
data: dict[str, str] = {"state": "closed"} response = self.client.patch(url, json=data)
response = client.patch(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return PullRequestModel(**response.json())
return PullRequestModel(**response.json())
def get_issue_comments(self, owner: str, repo: str, issue_number: int) -> list[CommentModel]: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return [CommentModel(**item) for item in response.json()]
return [CommentModel(**item) for item in response.json()]
def get_pull_request_comments(self, owner: str, repo: str, pull_number: int) -> list[CommentModel]: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/comments" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return [CommentModel(**item) for item in response.json()]
return [CommentModel(**item) for item in response.json()]
def get_pull_request_diff(self, owner: str, repo: str, pull_number: int) -> str: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/diff" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return response.text
return response.text
def get_pull_request_patch(self, owner: str, repo: str, pull_number: int) -> str: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/patch" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return response.text
return response.text
def get_pull_request_files(self, owner: str, repo: str, pull_number: int) -> list[PullRequestFileModel]: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/files" response = self.client.get(url)
response = client.get(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return [PullRequestFileModel(**item) for item in response.json()]
return [PullRequestFileModel(**item) for item in response.json()]
def list_assigned_issues(self, owner: str = "", repo: str = "") -> list[IssueModel]: def list_assigned_issues(self, owner: str = "", repo: str = "") -> list[IssueModel]:
try: try:
@@ -153,9 +157,8 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
return [] return []
username: str = user.login username: str = user.login
if owner and repo: if owner and repo:
response = httpx.get( response = self.client.get(
f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?assignee={username}&state=open&type=issues", f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?assignee={username}&state=open&type=issues",
headers=self.headers,
) )
response.raise_for_status() response.raise_for_status()
return [IssueModel(**item) for item in response.json()] return [IssueModel(**item) for item in response.json()]
@@ -164,9 +167,8 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
for r in repos: for r in repos:
repo_owner = r.owner repo_owner = r.owner
repo_name = r.name repo_name = r.name
resp = httpx.get( resp = self.client.get(
f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/issues?assignee={username}&state=open&type=issues", 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: if resp.status_code == 200:
for item in resp.json(): for item in resp.json():
@@ -188,9 +190,8 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
return [] return []
username: str = user.login username: str = user.login
if owner and repo: if owner and repo:
response = httpx.get( response = self.client.get(
f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state=open", f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state=open",
headers=self.headers,
) )
if response.status_code == 404: if response.status_code == 404:
return [] return []
@@ -205,9 +206,8 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
for r in repos: for r in repos:
repo_owner = r.owner repo_owner = r.owner
repo_name = r.name repo_name = r.name
resp = httpx.get( resp = self.client.get(
f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/pulls?state=open", f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/pulls?state=open",
headers=self.headers,
) )
if resp.status_code == 200: if resp.status_code == 200:
for pr_data in resp.json(): for pr_data in resp.json():
@@ -226,17 +226,16 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
self, owner: str, repo: str, head: str, base: str, title: str, description: str = "" self, owner: str, repo: str, head: str, base: str, title: str, description: str = ""
) -> PullRequestModel: ) -> PullRequestModel:
try: try:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls" data: dict[str, str] = {
data: dict[str, str] = { "title": title,
"title": title, "body": description,
"body": description, "head": head,
"head": head, "base": base,
"base": base, }
} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return PullRequestModel(**response.json())
return PullRequestModel(**response.json())
except Exception as e: except Exception as e:
logger.error(f"Error creating pull request: {e}", exc_info=True) logger.error(f"Error creating pull request: {e}", exc_info=True)
raise raise
@@ -251,18 +250,17 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
state: str | None = None, state: str | None = None,
) -> PullRequestModel: ) -> PullRequestModel:
try: try:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" data: dict[str, Any] = {}
data: dict[str, Any] = {} if title is not None:
if title is not None: data["title"] = title
data["title"] = title if body is not None:
if body is not None: data["body"] = body
data["body"] = body if state is not None:
if state is not None: data["state"] = state
data["state"] = state response = self.client.patch(url, json=data)
response = client.patch(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return PullRequestModel(**response.json())
return PullRequestModel(**response.json())
except Exception as e: except Exception as e:
logger.error(f"Error updating pull request: {e}", exc_info=True) logger.error(f"Error updating pull request: {e}", exc_info=True)
raise raise
@@ -273,47 +271,42 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
return self.create_pull_request(owner, repo, head, base, title, description) 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]: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" data: dict[str, Any] = {"event": "APPROVED", "body": comment}
data: dict[str, Any] = {"event": "APPROVED", "body": comment} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return response.json()
return response.json()
def request_changes_pr(self, owner: str, repo: str, pr_number: int, comment: str) -> dict[str, Any]: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment}
data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return response.json()
return response.json()
def get_pr_reviews(self, owner: str, repo: str, pr_number: int) -> list[dict[str, Any]]: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" response = self.client.get(url)
response = client.get(url, headers=self.headers) if response.status_code == 404:
if response.status_code == 404: return []
return [] response.raise_for_status()
response.raise_for_status() return response.json()
return response.json()
def dismiss_review_pr( def dismiss_review_pr(
self, owner: str, repo: str, pr_number: int, review_id: int, message: str self, owner: str, repo: str, pr_number: int, review_id: int, message: str
) -> dict[str, Any]: ) -> 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews/{review_id}/dismissals" data: dict[str, str] = {"message": message}
data: dict[str, str] = {"message": message} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return response.json()
return response.json()
def assign_issue(self, owner: str, repo: str, issue_number: int, username: str) -> IssueModel: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" data: dict[str, list[str]] = {"assignees": [username]}
data: dict[str, list[str]] = {"assignees": [username]} response = self.client.patch(url, json=data)
response = client.patch(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return IssueModel(**response.json())
return IssueModel(**response.json())
def create_issue( def create_issue(
self, self,
@@ -324,113 +317,103 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
labels: list[str] | None = None, labels: list[str] | None = None,
assignees: list[str] | None = None, assignees: list[str] | None = None,
) -> IssueModel: ) -> IssueModel:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues" data: dict[str, Any] = {"title": title, "body": body}
data: dict[str, Any] = {"title": title, "body": body} if labels:
if labels: data["labels"] = labels
data["labels"] = labels if assignees:
if assignees: data["assignees"] = assignees
data["assignees"] = assignees response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return IssueModel(**response.json())
return IssueModel(**response.json())
def add_comment(self, owner: str, repo: str, issue_number: int, body: str) -> CommentModel: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" data: dict[str, str] = {"body": body}
data: dict[str, str] = {"body": body} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return CommentModel(**response.json())
return CommentModel(**response.json())
def add_label(self, owner: str, repo: str, issue_number: int, label: str) -> LabelModel: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/labels" data: list[str] = [label]
data: list[str] = [label] response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return LabelModel(**response.json())
return LabelModel(**response.json())
def add_label_pr(self, owner: str, repo: str, pr_number: int, label: str) -> LabelModel: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pr_number}/labels" data: list[str] = [label]
data: list[str] = [label] response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return LabelModel(**response.json())
return LabelModel(**response.json())
def update_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/ref/{ref}" data: dict[str, str] = {"sha": sha}
data: dict[str, str] = {"sha": sha} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return response.json()
return response.json()
def create_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]: 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"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/refs" data: dict[str, str] = {"ref": ref, "sha": sha}
data: dict[str, str] = {"ref": ref, "sha": sha} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return response.json()
return response.json()
def update_file( def update_file(
self, owner: str, repo: str, path: str, message: str, content: str, branch: str self, owner: str, repo: str, path: str, message: str, content: str, branch: str
) -> dict[str, Any]: ) -> dict[str, Any]:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}" data: dict[str, str] = {
data: dict[str, str] = { "message": message,
"message": message, "content": base64.b64encode(content.encode()).decode(),
"content": base64.b64encode(content.encode()).decode(), "branch": branch,
"branch": branch, "new_branch": f"{branch}-update-{path}",
"new_branch": f"{branch}-update-{path}", }
} response = self.client.put(url, json=data)
response = client.put(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return response.json()
return response.json()
def get_file_content(self, owner: str, repo: str, path: str, ref: str = "master") -> str | list[str]: 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}"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}" params: dict[str, str] = {"ref": ref}
params: dict[str, str] = {"ref": ref} response = self.client.get(url, params=params)
response = client.get(url, headers=self.headers, params=params) response.raise_for_status()
response.raise_for_status() data = response.json()
data = response.json() if isinstance(data, list):
if isinstance(data, list): return [item.get("content", "") for item in data if item.get("type") == "file"]
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 ""
return base64.b64decode(data.get("content", "")).decode() if data.get("content") else ""
def list_unread_notifications(self, since: Optional[str] = None) -> list[dict[str, Any]]: def list_unread_notifications(self, since: Optional[str] = None) -> list[dict[str, Any]]:
try: try:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/notifications"
url = f"{self.base_url}/api/v1/notifications" params: dict[str, str] = {"all": "false"}
params: dict[str, str] = {"all": "false"} if since:
if since: params["since"] = since
params["since"] = since response = self.client.get(url, params=params)
response = client.get(url, headers=self.headers, params=params) response.raise_for_status()
response.raise_for_status() notifications: list[dict[str, Any]] = response.json()
notifications: list[dict[str, Any]] = response.json()
result: list[dict[str, Any]] = []
result: list[dict[str, Any]] = [] for n in notifications:
for n in notifications: repo_info = n.get("repository") or {}
repo_info = n.get("repository") or {} owner_info = repo_info.get("owner") or {}
owner_info = repo_info.get("owner") or {} owner_login = owner_info.get("login", "")
owner_login = owner_info.get("login", "") if owner_login == "meeks":
if owner_login == "meeks": result.append(n)
result.append(n) return result
return result
except Exception as e: except Exception as e:
logger.error(f"Error listing unread notifications: {e}", exc_info=True) logger.error(f"Error listing unread notifications: {e}", exc_info=True)
return [] return []
def mark_notification_as_read(self, thread_id: int) -> bool: def mark_notification_as_read(self, thread_id: int) -> bool:
try: try:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/notifications/threads/{thread_id}"
url = f"{self.base_url}/api/v1/notifications/threads/{thread_id}" response = self.client.patch(url)
response = client.patch(url, headers=self.headers) response.raise_for_status()
response.raise_for_status() return True
return True
except Exception as e: except Exception as e:
logger.error(f"Error marking notification thread {thread_id} as read: {e}", exc_info=True) logger.error(f"Error marking notification thread {thread_id} as read: {e}", exc_info=True)
return False return False
@@ -439,16 +422,15 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
self, owner: str, repo: str, pull_number: int, style: str = "squash", title: str = "", message: str = "" self, owner: str, repo: str, pull_number: int, style: str = "squash", title: str = "", message: str = ""
) -> bool: ) -> bool:
try: try:
with httpx.Client() as client: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/merge"
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/merge" data: dict[str, Any] = {
data: dict[str, Any] = { "Do": style,
"Do": style, "MergeTitleField": title,
"MergeTitleField": title, "MergeMessageField": message,
"MergeMessageField": message, }
} response = self.client.post(url, json=data)
response = client.post(url, headers=self.headers, json=data) response.raise_for_status()
response.raise_for_status() return True
return True
except Exception as e: except Exception as e:
logger.error(f"Error merging pull request {pull_number}: {e}", exc_info=True) logger.error(f"Error merging pull request {pull_number}: {e}", exc_info=True)
raise raise
tests/test_client.py
+2 -2
View File
@@ -56,7 +56,7 @@ def test_gitea_client_list_assigned_issues() -> None:
user_mock.login = "testuser" user_mock.login = "testuser"
with patch.object(client, "get_authenticated_user", return_value=user_mock), \ with patch.object(client, "get_authenticated_user", return_value=user_mock), \
patch("httpx.get") as mock_get: patch("httpx.Client.get") as mock_get:
mock_response: MagicMock = MagicMock() mock_response: MagicMock = MagicMock()
mock_response.status_code = 200 mock_response.status_code = 200
mock_response.json.return_value = [] mock_response.json.return_value = []
@@ -75,7 +75,7 @@ def test_gitea_client_list_assigned_pull_requests() -> None:
user_mock.login = "testuser" user_mock.login = "testuser"
with patch.object(client, "get_authenticated_user", return_value=user_mock), \ with patch.object(client, "get_authenticated_user", return_value=user_mock), \
patch("httpx.get") as mock_get: patch("httpx.Client.get") as mock_get:
mock_response: MagicMock = MagicMock() mock_response: MagicMock = MagicMock()
mock_response.status_code = 200 mock_response.status_code = 200
mock_response.json.return_value = [ mock_response.json.return_value = [