functions.py•24.5 kB
import gitlab
# 导入gitlab模块和配置函数
from src.gitlab_mcp_server import get_gitlab_client
# 实现直接使用GitLab API的函数
def create_repository(name, description=None, visibility="private", initialize_with_readme=True):
try:
gl = get_gitlab_client()
project = gl.projects.create({
'name': name,
'description': description,
'visibility': visibility,
'initialize_with_readme': initialize_with_readme
})
return {
"status": "success",
"message": f"Create {name} project successfully",
"id": project.id,
"name": project.name,
"web_url": project.web_url
}
except Exception as e:
return {
"status": "error",
"message": f"Create {name} project failed: {str(e)}"
}
def fork_repository(project_id, name=None, path=None, namespace_id=None):
"""Fork GitLab项目/仓库"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Fork {project_id} project failed: {str(e)}"
}
try:
# fork params
fork_params = {}
if name is not None:
fork_params['name'] = name
if path is not None:
fork_params['path'] = path
if namespace_id is not None:
fork_params['namespace_id'] = namespace_id
# Create fork
forked_project = project.forks.create(fork_params)
return {
"status": "success",
"message": f"Fork {project_id} project successfully",
"id": forked_project.id,
"name": forked_project.name,
"web_url": forked_project.web_url,
"forked_from_id": project_id
}
except Exception as e:
return {
"status": "error",
"message": f"Fork {project_id} project failed: {str(e)}"
}
def delete_repository(project_id):
"""删除GitLab项目/仓库"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {project_id} project failed: {str(e)}"
}
try:
# Delete the project
project.delete()
return {
"status": "success",
"message": f"project {project_id} successfully deleted"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {project_id} project failed: {str(e)}"
}
def search_repositories(search, page=1, per_page=20):
try:
gl = get_gitlab_client()
projects = gl.projects.list(search=search, page=page, per_page=per_page)
return {
"status": "success",
"message": f"Search {search} project successfully",
"projects": [{"id": p.id, "name": p.name, "path": p.path_with_namespace} for p in projects]
}
except Exception as e:
return {
"status": "error",
"message": f"Search {search} project failed: {str(e)}"
}
def create_or_update_file(project_id, file_path, content, commit_message, branch, ref_branch="master"):
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create or update file in {project_id} failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
# check if branch exists, if not create it
try:
# get the branch, if it exists return existing branch info
project.branches.get(branch)
except gitlab.exceptions.GitlabGetError:
# branch not found, create new branch
# ref_branch is the base branch to create new branch from
try:
project.branches.create({
'branch': branch,
'ref': ref_branch
})
except Exception as e:
return {
"status": "error",
"message": f"Create {branch} branch in {project_id} failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Create or update file in {branch} branch in {project_id} failed: {str(e)}"
}
try:
# Try to update existing file
f = project.files.get(file_path=file_path, ref=branch)
f.content = content
f.save(branch=branch, commit_message=commit_message)
return {
"status": "success",
"message": f"File {file_path} existing in {branch} branch in {project_id}",
"file_path": file_path
}
except gitlab.exceptions.GitlabGetError:
# Create new file
project.files.create({
'file_path': file_path,
'branch': branch,
'content': content,
'commit_message': commit_message
})
return {
"status": "success",
"message": f"File {file_path} created in {branch} branch in {project_id}",
"file_path": file_path
}
except Exception as e:
return {
"status": "error",
"message": f"Create or update file in {branch} branch in {project_id} failed: {str(e)}"
}
def push_files(project_id, files, commit_message, branch, ref_branch="master"):
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
# check if branch exists, if not create it
try:
# Try to get the branch, if it exists return existing branch info
project.branches.get(branch)
except gitlab.exceptions.GitlabGetError:
# If branch doesn't exist, create it base on ref_branch
try:
project.branches.create({
'branch': branch,
'ref': ref_branch
})
except Exception as e:
return {
"status": "error",
"message": f"Create {branch} branch base {ref_branch} branch in {project_id} project failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
# Prepare actions for commit
actions = []
for file_data in files:
file_path = file_data.get('file_path')
content = file_data.get('content', '')
action = file_data.get('action', 'create')
if action == 'delete':
actions.append({
'action': 'delete',
'file_path': file_path
})
else:
# For create/update, first check if file exists to determine action
try:
project.files.get(file_path=file_path, ref=branch)
# File exists, so we update
actions.append({
'action': 'update',
'file_path': file_path,
'content': content
})
except gitlab.exceptions.GitlabGetError:
# File doesn't exist, so we create
actions.append({
'action': 'create',
'file_path': file_path,
'content': content
})
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
# Create commit with all actions
try:
commit = project.commits.create({
'branch': branch,
'commit_message': commit_message,
'actions': actions
})
return {
"status": "success",
"message": f"Files pushed to {branch} branch of {project_id} project successfully",
"commit_id": commit.id,
"commit_short_id": commit.short_id,
"files_count": len(actions)
}
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
def get_file_contents(project_id, file_path, ref=None):
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} file contents failed: {str(e)}"
}
try:
f = project.files.get(file_path=file_path, ref=ref or project.default_branch)
return {
"status": "success",
"message": f"Get {file_path} file contents successfully",
"content": f.decode().decode()
}
except gitlab.exceptions.GitlabGetError:
return {
"error": "error",
"message": f"File {file_path} not found in {project_id}"
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} file contents failed: {str(e)}"
}
def create_issue(project_id, title, description=None, labels=None):
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {title} issue in {project_id} project failed: {str(e)}"
}
try:
issue = project.issues.create({
'title': title,
'description': description,
'labels': labels or []
})
return {
"status": "success",
"message": f"Create {title} issue in {project_id} project successfully",
"id": issue.iid,
"web_url": issue.web_url,
"state": issue.state
}
except Exception as e:
return {
"status": "error",
"message": f"Create {title} issue in {project_id} project failed: {str(e)}"
}
def get_issues(project_id, state=None, labels=None, page=1, per_page=20):
"""获取项目的所有issues，参考create_issue函数的实现结构"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} issues failed: {str(e)}"
}
# Prepare query parameters
query_params = {
'page': page,
'per_page': per_page
}
# Add optional filtering conditions
if state is not None:
query_params['state'] = state # opened, closed, all
if labels is not None:
if isinstance(labels, list):
query_params['labels'] = ','.join(labels)
else:
query_params['labels'] = labels
try:
# Get issues list
issues = project.issues.list(**query_params)
# Format return result, referencing create_issue's return format
formatted_issues = []
for issue in issues:
formatted_issues.append({
"id": issue.iid,
"title": issue.title,
"description": getattr(issue, 'description', ''),
"state": issue.state,
"labels": getattr(issue, 'labels', []),
"web_url": issue.web_url,
"author": getattr(issue, 'author', {}).get('name', 'Unknown'),
"created_at": getattr(issue, 'created_at', ''),
"updated_at": getattr(issue, 'updated_at', '')
})
return {
"status": "success",
"message": f"Get issues in {project_id} project successfully",
"issues": formatted_issues,
"total_count": len(formatted_issues),
"page": page,
"per_page": per_page
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} issues failed: {str(e)}"
}
def create_merge_request(project_id, title, source_branch, target_branch, description=None, draft=False):
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {project_id} merge request failed: {str(e)}"
}
# Check if merge request already exists
try:
existing_mrs = project.mergerequests.list(
source_branch=source_branch,
target_branch=target_branch,
state='opened'
)
if len(existing_mrs) > 0:
# Return the first matching merge request
existing_mr = existing_mrs[0]
return {
"status": "success",
"message": f"Merge request already exists from {source_branch} to {target_branch} in {project_id} project",
"id": existing_mr.iid,
"web_url": existing_mr.web_url,
"state": existing_mr.state
}
except Exception as e:
return {
"status": "error",
"message": f"merge request in {project_id} project failed: {str(e)}"
}
payload = {
'title': title,
'source_branch': source_branch,
'target_branch': target_branch,
'draft': draft
}
if description is not None:
payload['description'] = description
try:
mr = project.mergerequests.create(payload)
return {
"status": "success",
"message": f"Create merge request in {project_id} project successfully",
"id": mr.iid,
"web_url": mr.web_url,
"state": mr.state
}
except Exception as e:
return {
"status": "error",
"message": f"Create {project_id} merge request failed: {str(e)}"
}
def get_merge_request_diff(project_id, merge_request_iid):
"""获取GitLab合并请求的差异信息，用于查找有效的行位置进行评论"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"get {project_id} merge request diff failed failed: {str(e)}"
}
try:
# get the merge request by merge_request_iid
mr = project.mergerequests.get(merge_request_iid)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"merge_request {merge_request_iid} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"get {merge_request_iid} merge request diff failed failed: {str(e)}"
}
try:
# Get the diff refs
diff_refs = mr.diff_refs
# Fetch the changes
changes = mr.changes()
# Process the changes to include necessary information
diffs = []
for change in changes['changes']:
diff = {
'old_path': change['old_path'],
'new_path': change['new_path'],
'diff_refs': diff_refs,
'diff': change['diff']
}
diffs.append(diff)
return {
"status": "success",
"message": f"Get {merge_request_iid} merge request diff in {project_id} project successfully",
"diffs": diffs,
"merge_request_iid": merge_request_iid,
"project_id": project_id
}
except Exception as e:
return {
"status": "error",
"message": f"get {merge_request_iid} merge request diff failed failed: {str(e)}"
}
def create_branches(project_id, branch_name, ref_branch="master"):
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {branch_name} in {project_id} project failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
try:
# Try to get the branch, if it exists return existing branch info
existing_branch = project.branches.get(branch_name)
return {
"status": "success",
"message": f"Branch {branch_name} already exists in {project_id} project",
"branch_name": existing_branch.name,
"commit": existing_branch.commit
}
except gitlab.exceptions.GitlabGetError:
# Branch doesn't exist, create new branch
try:
new_branch = project.branches.create({
'branch': branch_name,
'ref': ref_branch
})
return {
"status": "success",
"message": f"Create {branch_name} in {project_id} project successfully",
"branch_name": new_branch.name,
"commit": new_branch.commit
}
except Exception as e:
return {
"status": "error",
"message": f"Create {branch_name} in {project_id} project failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {branch_name} in {project_id} project failed: {str(e)}"
}
def delete_branches(project_id, branch_name):
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {branch_name} in {project_id} project failed: {str(e)}"
}
# Check if it's the default branch or protected branch
if branch_name == project.default_branch:
return {
"status": "error",
"message": f"Cannot delete default branch: {branch_name} in {project_id} project"
}
# Delete the branch
try:
# Try to get the branch, if it exists then delete it
existing_branch = project.branches.get(branch_name)
existing_branch.delete()
return {
"status": "success",
"message": f"Branch {branch_name} in {project_id} project successfully deleted"
}
except gitlab.exceptions.GitlabGetError:
# Branch doesn't exist, which means deletion goal is already achieved
return {
"status": "success",
"message": f"Branch {branch_name} does not exist in {project_id} project (already deleted or never existed)"
}
except Exception as e:
return {
"status": "error",
"message": f"Failed to delete {branch_name} branch in {project_id} project: {str(e)}"
}
def create_tags(project_id, tag_name, ref_branch="master", message=None):
"""创建GitLab项目标签"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {tag_name} tag in {project_id} project failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
try:
# Try to get the tag, if it exists return existing tag info
existing_tag = project.tags.get(tag_name)
return {
"status": "success",
"message": f"Tag {tag_name} already exists in {project_id} project",
"tag_name": existing_tag.name,
"commit": existing_tag.commit,
"message": getattr(existing_tag, 'message', '')
}
except gitlab.exceptions.GitlabGetError:
# Tag doesn't exist, create new tag
try:
tag_params = {
'tag_name': tag_name,
'ref': ref_branch
}
if message is not None:
tag_params['message'] = message
new_tag = project.tags.create(tag_params)
return {
"status": "success",
"message": f"Create {tag_name} tag in {project_id} project successfully",
"tag_name": new_tag.name,
"commit": new_tag.commit,
"message": getattr(new_tag, 'message', '')
}
except Exception as e:
return {
"status": "error",
"message": f"Create {tag_name} tag in {project_id} project failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {tag_name} tag in {project_id} project failed: {str(e)}"
}
def delete_tags(project_id, tag_name):
"""删除GitLab项目标签"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {tag_name} tag in {project_id} project failed: {str(e)}"
}
try:
# Try to get the tag, if it exists then delete it
existing_tag = project.tags.get(tag_name)
# Delete the tag
existing_tag.delete()
return {
"status": "success",
"message": f"delete {tag_name} tag successfully"
}
except gitlab.exceptions.GitlabGetError:
return {
"status": "success",
"message": f"Tag {tag_name} does not exist in {project_id} project (already deleted or never existed)"
}
except Exception as e:
# Tag doesn't exist
return {
"status": "error",
"message": f"Delete {tag_name} tag in {project_id} project failed: {str(e)}"
}