"""
GitCode 仓库API
"""
from typing import Any, Dict, List, Optional, Union
from ..base.client import BaseAPI
class RepositoryAPI(BaseAPI):
"""仓库API"""
async def list_user_repos(self, visibility: Optional[str] = None, sort: Optional[str] = None) -> List[Dict[str, Any]]:
"""
获取当前用户的仓库列表
Args:
visibility: 仓库可见性过滤 ('all'|'public'|'private')
sort: 排序方式 ('created'|'updated'|'pushed'|'full_name')
Returns:
仓库列表
"""
params = {
"visibility": visibility,
"sort": sort,
}
return await self.request("GET", "/user/repos", params=params)
async def get_repo(self, owner: str, repo: str) -> Dict[str, Any]:
"""
获取特定仓库的详细信息
Args:
owner: 仓库所有者
repo: 仓库名称
Returns:
仓库信息
"""
return await self.request("GET", f"/repos/{owner}/{repo}")
async def create_repo(
self,
name: str,
description: str = "",
private: bool = False,
has_issues: bool = True,
has_wiki: bool = True,
auto_init: bool = False,
gitignore_template: Optional[str] = None,
license_template: Optional[str] = None
) -> Dict[str, Any]:
"""
创建新仓库
Args:
name: 仓库名称
description: 仓库描述
private: 是否为私有仓库
has_issues: 是否启用Issues
has_wiki: 是否启用Wiki
auto_init: 是否自动初始化仓库（创建README）
gitignore_template: gitignore模板名称
license_template: 许可证模板名称
Returns:
创建的仓库信息
"""
data = {
"name": name,
"description": description,
"private": private,
"has_issues": has_issues,
"has_wiki": has_wiki,
"auto_init": auto_init,
"gitignore_template": gitignore_template,
"license_template": license_template
}
return await self.request("POST", "/user/repos", data=data)
async def update_repo(
self,
owner: str,
repo: str,
name: Optional[str] = None,
description: Optional[str] = None,
private: Optional[bool] = None,
has_issues: Optional[bool] = None,
has_wiki: Optional[bool] = None,
default_branch: Optional[str] = None
) -> Dict[str, Any]:
"""
更新仓库信息
Args:
owner: 仓库所有者
repo: 仓库名称
name: 新的仓库名称
description: 新的仓库描述
private: 是否为私有仓库
has_issues: 是否启用Issues
has_wiki: 是否启用Wiki
default_branch: 默认分支
Returns:
更新后的仓库信息
"""
data = {k: v for k, v in {
"name": name,
"description": description,
"private": private,
"has_issues": has_issues,
"has_wiki": has_wiki,
"default_branch": default_branch
}.items() if v is not None}
return await self.request("PATCH", f"/repos/{owner}/{repo}", data=data)
async def delete_repo(self, owner: str, repo: str) -> None:
"""
删除仓库
Args:
owner: 仓库所有者
repo: 仓库名称
"""
await self.request("DELETE", f"/repos/{owner}/{repo}")
async def list_collaborators(self, owner: str, repo: str) -> List[Dict[str, Any]]:
"""
列出仓库协作者
Args:
owner: 仓库所有者
repo: 仓库名称
Returns:
协作者列表
"""
return await self.request("GET", f"/repos/{owner}/{repo}/collaborators")
async def add_collaborator(self, owner: str, repo: str, username: str, permission: str = "push") -> Dict[str, Any]:
"""
添加协作者
Args:
owner: 仓库所有者
repo: 仓库名称
username: 用户名
permission: 权限级别 ('pull'|'push'|'admin')
Returns:
邀请信息
"""
data = {"permission": permission}
return await self.request("PUT", f"/repos/{owner}/{repo}/collaborators/{username}", data=data)
async def remove_collaborator(self, owner: str, repo: str, username: str) -> None:
"""
移除协作者
Args:
owner: 仓库所有者
repo: 仓库名称
username: 用户名
"""
await self.request("DELETE", f"/repos/{owner}/{repo}/collaborators/{username}")
async def get_readme(self, owner: str, repo: str, ref: Optional[str] = None) -> Dict[str, Any]:
"""
获取仓库README
Args:
owner: 仓库所有者
repo: 仓库名称
ref: 分支或提交SHA
Returns:
README文件信息
"""
params = {"ref": ref} if ref else None
return await self.request("GET", f"/repos/{owner}/{repo}/readme", params=params)
async def get_contents(self, owner: str, repo: str, path: str, ref: Optional[str] = None) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
"""
获取仓库内容
Args:
owner: 仓库所有者
repo: 仓库名称
path: 文件或目录路径
ref: 分支或提交SHA
Returns:
文件内容或目录列表
"""
params = {"ref": ref} if ref else None
return await self.request("GET", f"/repos/{owner}/{repo}/contents/{path}", params=params)
async def create_file(
self,
owner: str,
repo: str,
path: str,
message: str,
content: str,
branch: Optional[str] = None,
committer: Optional[Dict[str, str]] = None,
author: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""
创建文件
Args:
owner: 仓库所有者
repo: 仓库名称
path: 文件路径
message: 提交消息
content: 文件内容（Base64编码）
branch: 分支名
committer: 提交者信息 {'name': '名称', 'email': '邮箱'}
author: 作者信息 {'name': '名称', 'email': '邮箱'}
Returns:
提交信息
"""
data = {
"message": message,
"content": content,
"branch": branch,
"committer": committer,
"author": author
}
data = {k: v for k, v in data.items() if v is not None}
return await self.request("PUT", f"/repos/{owner}/{repo}/contents/{path}", data=data)
async def update_file(
self,
owner: str,
repo: str,
path: str,
message: str,
content: str,
sha: str,
branch: Optional[str] = None,
committer: Optional[Dict[str, str]] = None,
author: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""
更新文件
Args:
owner: 仓库所有者
repo: 仓库名称
path: 文件路径
message: 提交消息
content: 文件内容（Base64编码）
sha: 文件的当前SHA
branch: 分支名
committer: 提交者信息 {'name': '名称', 'email': '邮箱'}
author: 作者信息 {'name': '名称', 'email': '邮箱'}
Returns:
提交信息
"""
data = {
"message": message,
"content": content,
"sha": sha,
"branch": branch,
"committer": committer,
"author": author
}
data = {k: v for k, v in data.items() if v is not None}
return await self.request("PUT", f"/repos/{owner}/{repo}/contents/{path}", data=data)
async def delete_file(
self,
owner: str,
repo: str,
path: str,
message: str,
sha: str,
branch: Optional[str] = None,
committer: Optional[Dict[str, str]] = None,
author: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""
删除文件
Args:
owner: 仓库所有者
repo: 仓库名称
path: 文件路径
message: 提交消息
sha: 文件的当前SHA
branch: 分支名
committer: 提交者信息 {'name': '名称', 'email': '邮箱'}
author: 作者信息 {'name': '名称', 'email': '邮箱'}
Returns:
提交信息
"""
data = {
"message": message,
"sha": sha,
"branch": branch,
"committer": committer,
"author": author
}
data = {k: v for k, v in data.items() if v is not None}
return await self.request("DELETE", f"/repos/{owner}/{repo}/contents/{path}", data=data)
