b2ad726fc4
Implemented full async Gitea REST API v1 client using httpx with the following methods: - create_pull_request: Create PRs with title, head, base, and body - merge_pull_request: Merge PRs with configurable merge type - create_issue_comment: Post comments on issues/PRs - get_issue: Fetch issue/PR details - get_issue_comments: Retrieve all comments for an issue/PR - create_branch: Create new branches from existing ones Added lazy singleton pattern with get_gitea_client() factory function that reads GITEA_URL and GITEA_TOKEN from environment. All methods properly call raise_for_status() and return JSON responses. Comprehensive test suite with 8 tests covering all methods plus error handling.
147 lines
4.2 KiB
Python
147 lines
4.2 KiB
Python
"""Gitea REST API v1 client. Phase 2 implementation."""
|
|
|
|
import os
|
|
import httpx
|
|
|
|
|
|
class GiteaClient:
|
|
def __init__(self, base_url: str, token: str):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.token = token
|
|
self._client = httpx.AsyncClient(
|
|
base_url=f"{self.base_url}/api/v1",
|
|
headers={"Authorization": f"token {self.token}"},
|
|
)
|
|
|
|
async def create_pull_request(self, owner, repo, title, head, base, body) -> dict:
|
|
"""Create a pull request.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
title: PR title
|
|
head: Head branch name
|
|
base: Base branch name
|
|
body: PR body/description
|
|
|
|
Returns:
|
|
dict: Created PR data (number, html_url, etc.)
|
|
"""
|
|
resp = await self._client.post(
|
|
f"/repos/{owner}/{repo}/pulls",
|
|
json={"title": title, "head": head, "base": base, "body": body},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def merge_pull_request(self, owner, repo, pr_number, merge_type="merge") -> dict:
|
|
"""Merge a pull request.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
pr_number: PR number
|
|
merge_type: Merge type ("merge", "rebase", "squash")
|
|
|
|
Returns:
|
|
dict: Merge result
|
|
"""
|
|
resp = await self._client.post(
|
|
f"/repos/{owner}/{repo}/pulls/{pr_number}/merge",
|
|
json={"Do": merge_type},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def create_issue_comment(self, owner, repo, issue_number, body) -> dict:
|
|
"""Create a comment on an issue or PR.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
issue_number: Issue or PR number
|
|
body: Comment body
|
|
|
|
Returns:
|
|
dict: Created comment data (id, body, etc.)
|
|
"""
|
|
resp = await self._client.post(
|
|
f"/repos/{owner}/{repo}/issues/{issue_number}/comments",
|
|
json={"body": body},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def get_issue(self, owner, repo, issue_number) -> dict:
|
|
"""Get issue or PR details.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
issue_number: Issue or PR number
|
|
|
|
Returns:
|
|
dict: Issue/PR data (number, title, body, etc.)
|
|
"""
|
|
resp = await self._client.get(f"/repos/{owner}/{repo}/issues/{issue_number}")
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def get_issue_comments(self, owner, repo, issue_number) -> list:
|
|
"""Get all comments on an issue or PR.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
issue_number: Issue or PR number
|
|
|
|
Returns:
|
|
list: List of comment dicts
|
|
"""
|
|
resp = await self._client.get(
|
|
f"/repos/{owner}/{repo}/issues/{issue_number}/comments"
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def create_branch(self, owner, repo, branch_name, old_branch) -> dict:
|
|
"""Create a new branch.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
branch_name: New branch name
|
|
old_branch: Source branch name
|
|
|
|
Returns:
|
|
dict: Created branch data
|
|
"""
|
|
resp = await self._client.post(
|
|
f"/repos/{owner}/{repo}/branches",
|
|
json={"new_branch_name": branch_name, "old_branch_name": old_branch},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def close(self):
|
|
await self._client.aclose()
|
|
|
|
|
|
# Lazy singleton
|
|
_client: GiteaClient | None = None
|
|
|
|
|
|
def get_gitea_client() -> GiteaClient:
|
|
"""Get or create the singleton GiteaClient instance.
|
|
|
|
Returns:
|
|
GiteaClient: The singleton instance
|
|
"""
|
|
global _client
|
|
if _client is None:
|
|
_client = GiteaClient(
|
|
base_url=os.environ.get("GITEA_URL", "http://gitea:3000"),
|
|
token=os.environ.get("GITEA_TOKEN", ""),
|
|
)
|
|
return _client
|