呼叫gitlibapi v3保護分支,並開通Developers can push
阿新 • • 發佈:2018-12-26
第三方包要求需要安裝requests
# -*- coding:utf-8 -*-
import requests, re, logging, sys
class GitLabAPI(object):
def __init__(self, headers=None, *args, **kwargs):
self.headers = headers
self.logger = logging.getLogger("GitLabAPI")
self.formatter = logging.Formatter('%(levelname)s: %(message)s' )
self.logger.setLevel(logging.INFO)
# 檔案日誌
self.file_handler = logging.FileHandler("GitLabAPI.log")
self.file_handler.setFormatter(self.formatter) # 可以通過setFormatter指定輸出格式
# 控制檯日誌
self.console_handler = logging.StreamHandler(sys.stdout)
self.console_handler.formatter = self.formatter # 也可以直接給formatter賦值
# 為logger新增的日誌處理器
self.logger.addHandler(self.file_handler)
self.logger.addHandler(self.console_handler)
'''
搜尋使用者可以看到的所有工程
:name: 工程名
:result: 工程集合
'''
def get_user_projects(self, name):
res = requests.get("http://example.com/api/v3/projects/all?search=%s&per_page=100" % name, headers=self.headers, verify=False)
status_code = res.status_code
if status_code != 200:
raise RuntimeError(res.json().get('message'))
content = res.json()
return content
'''
獲取使用者工程id
:param name: 專案名稱
:return: 工程id,工程訪問路徑
'''
def get_user_project_id(self, name):
project_id = None
web_url = None
projects = self.get_user_projects(name)
if projects:
for item in projects:
if item.get('name') == name:
project_id = item.get('id')
web_url = item.get('web_url')
if project_id is None:
raise RuntimeError('找不到工程:%s' % name)
return project_id, web_url
'''
獲取工程的所有分支
:project_id: 工程id
:return: 分支名陣列
'''
def get_project_branchs(self, project_id):
branchs = []
res = requests.get("http://example.com/api/v3/projects/%s/repository/branches"%project_id, headers=self.headers, verify=False)
status_code = res.status_code
if status_code != 200:
raise RuntimeError(res.json().get('message'))
content = res.json()
if content:
for item in content:
branchs.append(item.get('name'))
return branchs
'''
獲取工程的指定分支是否已經保護
:project_id: 工程id
:branch_name: 分支名
:return: True or False
'''
def get_project_branch(self, project_id, branch_name):
res = requests.get("http://example.com/api/v3/projects/%s/repository/branches/%s" % (project_id, branch_name), headers=self.headers, verify=False)
status_code = res.status_code
if status_code != 200:
raise RuntimeError(res.json().get('message'))
content = res.json()
if content:
return content.get('protected')
return False
'''
保護單個工程的分支
:session: requests session用於保持登入
:project_name: 工程名
:branch_name: 分支名
:result: True or False
'''
def project_protect_branch(self, session, project_name, branch_name):
self.logger.info("工程:%s 分支:%s" % (project_name, branch_name))
project_id, web_url = self.get_user_project_id(project_name)
self.logger.info("工程id:%s" % project_id)
branchs = self.get_project_branchs(project_id)
self.logger.debug("所有分支:%s" % branchs)
if branch_name in branchs:
content = self.get_project_branch(project_id, branch_name)
if content:
self.logger.info('工程:%s 分支:%s 已經被保護,不能重複保護' % (project_name, branch_name))
return True
protected_page = session.get(web_url+'/protected_branches')
authenticity_token = re.findall(r'<meta content="([^"]+)" name="csrf-token" />',protected_page.text)
protect_data = {'utf8':'✓','authenticity_token':authenticity_token,'protected_branch[name]':branch_name,
'protected_branch[developers_can_push]':0,'protected_branch[developers_can_push]':1}
session.post(web_url+'/protected_branches', protect_data)
content = self.get_project_branch(project_id, branch_name)
return content
else:
raise RuntimeError('工程:%s 找不到分支:%s' % (project_name, branch_name))
'''
保護多個工程的分支
:project_names: 工程名陣列
:branch_name: 分支名
:username: gitlib使用者名稱
:password: gitlib密碼
'''
def projects_protect_branch(self, project_names, branch_name, username, password):
session = requests.Session()
login_page = session.get('http://example.com/users/sign_in')
authenticity_token = re.findall(r'<meta content="([^"]+)" name="csrf-token" />',login_page.text)
#print(authenticity_token)
login_data = {'utf8':'✓','authenticity_token':authenticity_token,'user[login]':username,'user[password]':password,'user[remember_me]':0}
res = session.post('http://example.com/users/sign_in', data=login_data)
#print(res.text)
for project_name in project_names:
self.logger.info("-----------------------------------------------------------------------------------------------")
try:
result = api.project_protect_branch(session, project_name, branch_name)
if result:
self.logger.info("工程:%s 分支:%s 保護成功 message:%s" % (project_name, branch_name, result))
else:
self.logger.error("工程:%s 分支:%s 保護失敗 message:%s" % (project_name, branch_name, result))
except Exception as e:
self.logger.exception("工程:%s 分支:%s 保護失敗" % (project_name, branch_name))
self.logger.info("-----------------------------------------------------------------------------------------------")
if __name__ == "__main__":
headers = {'PRIVATE-TOKEN': 'privatetoken'} #你的gitlab賬戶Account裡面的private token
username = 'username'#你的gitlab賬戶的使用者名稱
password = 'password'#你的gitlab賬戶的密碼
branch_name = 'branch_name'#分支名
projects = ['project_name']#工程名陣列
api = GitLabAPI(headers=headers)
api.projects_protect_branch(projects, branch_name, username, password)