xiaxinhang/pr-agent
1
0
Fork 0
mirror of https://github.com/qodo-ai/pr-agent.git synced 2025-07-02 11:50:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement skip branches for github, add ignore title to gitlab

Browse Source
This commit is contained in:
Paolo Mainardi
2024-09-02 16:26:50 +02:00
parent 23af1afa03
commit adce35765b
3 changed files with 72 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
pr_agent/servers/gitlab_webhook.py
View File

@ -1,4 +1,5 @@
import copy
import re
import json
from datetime import datetime
@ -124,29 +125,48 @@ async def gitlab_webhook(background_tasks: BackgroundTasks, request: Request):
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
log_context["sender"] = sender
excluded_source_branches = get_settings().get("gitlab.excluded_source_branches", [])
excluded_target_branches = get_settings().get("gitlab.excluded_target_branches", [])
excluded_labels = get_settings().get("gitlab.excluded_labels", [])
if data.get('object_kind') == 'merge_request' and data['object_attributes'].get('action') in ['open', 'reopen']:
title = data['object_attributes'].get('title')
url = data['object_attributes'].get('url')
draft = data['object_attributes'].get('draft')
source_branch = data['object_attributes'].get('source_branch')
target_branch = data['object_attributes'].get('target_branch')
labels = [label['title'] for label in data['object_attributes'].get('labels', [])]
get_logger().info(f"New merge request: {url}")
if target_branch in excluded_target_branches or source_branch in excluded_source_branches:
get_logger().info(f"Skipping excluded branch MR: {url}")
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
if labels.intersection(excluded_labels):
get_logger().info(f"Skipping excluded label MR: {url}")
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
# ignore draft MRs.
if draft:
get_logger().info(f"Skipping draft MR: {url}")
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
# logic to ignore MRs for titles, labels and source, target branches.
ignore_mr_title = get_settings().get("GITLAB.IGNORE_MR_TITLE", [])
ignore_mr_labels = get_settings().get("GITLAB.IGNORE_MR_LABELS", [])
ignore_mr_source_branches = get_settings().get("GITLAB.IGNORE_MR_SOURCE_BRANCHES", [])
ignore_mr_target_branches = get_settings().get("GITLAB.IGNORE_MR_TARGET_BRANCHES", [])
if ignore_mr_source_branches:
source_branch = data['object_attributes'].get('source_branch')
if any(re.search(regex, source_branch) for regex in ignore_mr_source_branches):
get_logger().info(f"Ignoring MR with source branch '{source_branch}' due to gitlab.ignore_mr_source_branches settings")
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
if ignore_mr_target_branches:
target_branch = data['object_attributes'].get('target_branch')
if any(re.search(regex, target_branch) for regex in ignore_mr_target_branches):
get_logger().info(f"Ignoring MR with target branch '{target_branch}' due to gitlab.ignore_mr_target_branches settings")
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
if ignore_mr_labels:
labels = [label['title'] for label in data['object_attributes'].get('labels', [])]
if any(label in ignore_mr_labels for label in labels):
labels_str = ", ".join(labels)
get_logger().info(f"Ignoring MR with labels '{labels_str}' due to gitlab.ignore_mr_labels settings")
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
if ignore_mr_title:
if any([re.search(regex, title) for regex in ignore_mr_title]):
get_logger().info(f"Ignoring MR with title '{title}' due to gitlab.ignore_mr_title settings")
return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
await _perform_commands_gitlab("pr_commands", PRAgent(), url, log_context)
elif data.get('object_kind') == 'note' and data.get('event_type') == 'note': # comment on MR
if 'merge_request' in data: