meeks
ae4e2d46ac
- Replace Any with object or specific types across codebase - Add ReviewRequest dataclass for PR review payloads - Update bad_code.md: mark 5.1 (Any Type Overuse) as resolved - Fix summary table with accurate counts and unresolved issues list
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
"""Files client for Gitea API operations."""
|
|
|
|
import base64
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
|
|
logger: logging.Logger = logging.getLogger("gitea.files_client")
|
|
|
|
|
|
class FilesClient:
|
|
"""HTTP client for Gitea Files and Git Refs API operations."""
|
|
|
|
def __init__(self, base_url: str, client: httpx.Client) -> None:
|
|
"""Initialize the FilesClient.
|
|
|
|
Args:
|
|
base_url: The base URL for the Gitea API.
|
|
client: The httpx client for making requests.
|
|
"""
|
|
self.base_url: str = base_url
|
|
self.client: httpx.Client = client
|
|
|
|
def update_file(
|
|
self, owner: str, repo: str, path: str, message: str, content: str, branch: str
|
|
) -> dict[str, object]:
|
|
"""Update a file in a repository.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
path: File path.
|
|
message: Commit message.
|
|
content: File content.
|
|
branch: Branch name.
|
|
|
|
Returns:
|
|
The API response.
|
|
"""
|
|
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 = self.client.put(url, 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]:
|
|
"""Get the content of a file or directory.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
path: File or directory path.
|
|
ref: Git reference (branch, tag, commit).
|
|
|
|
Returns:
|
|
File content as string, or list of file names if path is a directory.
|
|
"""
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}"
|
|
params: dict[str, str] = {"ref": ref}
|
|
response = self.client.get(url, 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 ""
|
|
)
|
|
|
|
def update_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, object]:
|
|
"""Update a git reference.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
ref: Reference name (e.g., heads/main).
|
|
sha: New SHA for the reference.
|
|
|
|
Returns:
|
|
The API response.
|
|
"""
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/ref/{ref}"
|
|
data: dict[str, str] = {"sha": sha}
|
|
response = self.client.post(url, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def create_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, object]:
|
|
"""Create a new git reference.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
ref: Reference name (e.g., refs/heads/new-branch).
|
|
sha: SHA for the reference.
|
|
|
|
Returns:
|
|
The API response.
|
|
"""
|
|
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/refs"
|
|
data: dict[str, str] = {"ref": ref, "sha": sha}
|
|
response = self.client.post(url, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|