适用场景:批量拉取代码,仓库转移
一、环境信息
脚本适用于:python2
测试版本:2.7.18
二、参数说明
apiToken: gitlab个人账户授权的apitoken ,通过点击右上角【用户头像】->选择【Edit profile】 ->菜单中选择【 访问令牌】,右侧 进行创建,建议给予全部权限
projectUrl: gitlab地址,一般只需要修改ip和端口
三、脚本使用说明
1. 使用python2运行git.py
python git.py
2. python脚本执行完毕会自动生成如下四个文件
-
allproject.json: 所有有权限的非空项目信息
-
projectclone.sh : 所有项目组成的git clone 命令
-
projectclonemirror.sh 所有项目组成的git clone -mirror命令
-
projectpull.sh 所有项目组成的git pull命令
3. 其他脚本说明
- update.sh: 从目录向下扫描git、svn项目并更新
四、脚本源码
1. git.py
通过gitlab 提供的api接口访问项目信息
# coding: utf-8
import os
import requests
import json
import sys
#python2
reload(sys)
sys.setdefaultencoding('utf-8')
#配置信息,根据实际情况修改
apiToken="X******Seq-******tc"
projectUrl='http://127.0.0.1/api/v4/projects'
#页容量
pageSize=100
projectInfo=[]
currPath=os.getcwd()
#所有项目信息(路径根据所使用环境信息,可能需要修改分隔符)
projectJson=open(currPath+"\\allproject.json",mode="w")
#clone所有项目
projectClone=open(currPath+"\\projectclone.sh",mode="w")
#clone --mirror所有项目
projectCloneMirror=open(currPath+"\\projectclonemirror.sh",mode="w")
#pull所有分支
projectPull=open(currPath+"\\projectpull.sh",mode="w")
projectClone.write("#!/bin/bash\nworkpath=$(cd \"$(dirname \"$0\")\";pwd)\n")
projectCloneMirror.write("#!/bin/bash\nworkpath=$(cd \"$(dirname \"$0\")\";pwd)\n")
projectPull.write("#!/bin/bash\nworkpath=$(cd \"$(dirname \"$0\")\";pwd)\n")
#get请求gitlab的v4 api接口
def get(getUrl):
pageNo = 1
totalPage = 1
totaldata = [];
while pageNo < totalPage+1:
res = requests.get(url=getUrl+'?per_page='+str(pageSize)+'&page='+str(pageNo),headers={"PRIVATE-TOKEN":apiToken})
#print(res.headers)
totalPage=int(res.headers.get('X-Total-Pages'))
totaldata.extend(json.loads(res.text));
#print('total:',totalPage,'pageNo',pageNo)
pageNo+=1
#print(len(totaldata))
return totaldata;
#获取所有分支
def getBranch(branchRepo):
branchR = []
branchs = get(branchRepo)
for branch in branchs:
branchR.append(branch['name'])
return branchR
#获取所有tag
def getTages(branchRepo):
tagRepo = branchRepo.replace("/branches","/tags")
branchR = []
branchs = get(tagRepo)
for branch in branchs:
branchR.append(branch['name'])
return branchR
#获取所有项目
def getAllProject():
projectList = get(projectUrl)
for proj in projectList:
if proj['empty_repo']:
print('null project:',proj['path_with_namespace'])
else:
projInfo = {}
projInfo['id'] = proj['id']
projInfo['path_with_namespace'] = proj['path_with_namespace']
projInfo['empty_repo'] = proj['empty_repo']
projInfo['http_url_to_repo'] = proj['http_url_to_repo']
projInfo['default_branch'] = proj['default_branch']
projInfo['repo_branches'] = proj['_links']['repo_branches']
projInfo['branchs'] = getBranch(projInfo['repo_branches'])
projInfo['tages'] = getTages(projInfo['repo_branches'])
projectInfo.append(projInfo)
#处理输出文件,可单独处理
#处理clone脚本
projectClone.write("cd $workpath\n")
projectClone.write("git clone " + projInfo['http_url_to_repo'] + " " + projInfo['path_with_namespace'] +" \n")
projectClone.write("cd "+projInfo['path_with_namespace']+"\n")
#处理clone --mirror脚本
projectCloneMirror.write("cd $workpath\n")
projectCloneMirror.write("git clone --mirror " + projInfo['http_url_to_repo'] + " " + projInfo['path_with_namespace'] +" \n")
projectCloneMirror.write("cd "+projInfo['path_with_namespace']+"\n")
#处理pull脚本
projectPull.write("cd $workpath\n")
projectPull.write("cd "+projInfo['path_with_namespace']+"\n")
for branch in projInfo['branchs']:
projectClone.write("git checkout \""+branch+"\"\n")
projectClone.write("git pull\n")
projectCloneMirror.write("git checkout \""+branch+"\"\n")
projectCloneMirror.write("git pull\n")
projectPull.write("git checkout \""+branch+"\"\n")
projectPull.write("git pull\n")
for tag in projInfo['tages']:
projectClone.write("git checkout \""+tag+"\"\n")
projectCloneMirror.write("git checkout \""+tag+"\"\n")
projectPull.write("git checkout \""+tag+"\"\n")
#projectCloneMirror.write("#!/bin/bash\nworkpath=$(cd \"$(dirname \"$0\")\";pwd)\n")
#projectPull.write("#!/bin/bash\nworkpath=$(cd \"$(dirname \"$0\")\";pwd)\n")
return projectInfo
#
getAllProject()
#将所有项目信息写入文件
projectJson.write(json.dumps(projectInfo))
projectJson.flush()
projectJson.close()
projectClone.flush()
projectClone.close()
projectCloneMirror.flush()
projectCloneMirror.close()
projectPull.flush()
projectPull.close()
#直接调起系统命令
print(os.system('git --version'))
2. update.sh
#自动识别仓库类型(git/svn)并更新
#!/bin/bash
basepath=$(pwd)
echo $basepath
updateSource(){
#local currpath=$1
cd $1
if [ -d $1/.git ]; then
echo 'start update sourcecode'$1
echo "开始更新"$currpath
git pull
#老版本的svn每个文件夹一个.svn文件夹,
elif [ -d $1/.svn ]; then
#revision=`svn info |grep "Last Changed Rev:" |awk '{print $4}'`
revision=`svn info |grep "最后修改的版本" |awk '{print $2}'`
if [ $revision > 0 ]; then
echo '$revision 开始更新代码'$1
svn update .
fi
else
for childfile in `ls .`
do
pwd
if [ -d $1/$childfile ]; then
updateSource $1/$childfile
else
echo $1/$childfile' is not dir , will ignore-------<'
fi
done
fi
}
#main run
updateSource $basepath
五、仓库迁移
通过api创建分组、项目
通过git clone --mirror 及 git push --mirror 迁移仓库
接口异常未处理!文章来源:https://www.toymoban.com/news/detail-430144.html
# coding: utf-8
import os
import requests
import json
import sys
import time
#python2
reload(sys)
sys.setdefaultencoding('utf-8')
#源库信息
apiToken="****-******"
projectUrl='http://127.0.0.1:1307/api/v4/projects'
#目标库信息
targetapiToken="****-******"
targetHostUrl='http://127.0.0.1:1306/api/v4/'
#目标库分组id(将按源目录自动提交到此分组下)
targetNamespaceId=6
#目标库分组名称
targetNamespaceName="demo/"
#目标库删除异步等待时间(重新创建)
deleteWaitsec=5
pageSize=100
projectInfo=[]
currPath=os.getcwd()
#所有项目信息
projectJson=open(currPath+"\\allproject.json",mode="w")
#clone --mirror所有项目
projectCloneMirror=open(currPath+"\\projectclonemirrorpush.sh",mode="w")
projectCloneMirror.write("#!/bin/bash\nworkpath=$(cd \"$(dirname \"$0\")\";pwd)\n")
#get请求gitlab的v4 api接口
def get(getUrl):
pageNo = 1
totalPage = 1
totaldata = [];
while pageNo < totalPage+1:
res = requests.get(url=getUrl+'?per_page='+str(pageSize)+'&page='+str(pageNo),headers={"PRIVATE-TOKEN":apiToken})
#print(res.headers)
totalPage=int(res.headers.get('X-Total-Pages'))
totaldata.extend(json.loads(res.text));
#print('total:',totalPage,'pageNo',pageNo)
pageNo+=1
#print(len(totaldata))
return totaldata;
#获取所有分支
def getBranch(branchRepo):
branchR = []
branchs = get(branchRepo)
for branch in branchs:
branchR.append(branch['name'])
return branchR
#获取所有tag
def getTages(branchRepo):
tagRepo = branchRepo.replace("/branches","/tags")
branchR = []
branchs = get(tagRepo)
for branch in branchs:
branchR.append(branch['name'])
return branchR
def createGroup(pgroupId,groupName):
print("startcreateGroup",pgroupId,groupName)
res = requests.get(url=targetHostUrl+'groups?id='+str(pgroupId)+'&search='+groupName,headers={"PRIVATE-TOKEN":targetapiToken})
if res.status_code<300 and res.status_code >= 199:
#请求成功
resdata = (json.loads(res.text));
if len(resdata) > 0:
#存在,返回groupid
print("hasgroup",pgroupId,groupName)
return resdata[0]['id']
else:
#不存在,创建
print("newgroup",pgroupId,groupName)
reqdata={"name":groupName,"path":groupName,"parent_id":pgroupId}
res = requests.post(url=targetHostUrl+'groups',data=reqdata,headers={"PRIVATE-TOKEN":targetapiToken})
resdata = (json.loads(res.text));
return resdata['id']
else:
print("errorgroup",pgroupId,groupName)
def pushOnGroup(prowithNamespace,groupId):
#验证项目是否存在
res = requests.get(url=targetHostUrl+'projects?search='+targetNamespaceName+prowithNamespace+'&search_namespaces=true',headers={"PRIVATE-TOKEN":targetapiToken})
resdata = (json.loads(res.text));
if len(resdata) > 0:
#项目存在
print("projecthas",prowithNamespace,groupId)
proId = resdata[0]["id"]
requests.delete(url=targetHostUrl+'projects/'+str(proId),headers={"PRIVATE-TOKEN":targetapiToken})
time.sleep(deleteWaitsec)
grouppros = prowithNamespace.split("/")
currgroupId = groupId;
proName=""
for index in range(len(grouppros)):
#最后一层是项目名
if index == len(grouppros)-1:
proName = grouppros[index]
else:
currgroupId = createGroup(currgroupId,grouppros[index])
#创建项目
print("startCreateProject",proName,currgroupId)
prodata={"name": proName,"description":proName,"path": proName,"namespace_id": str(currgroupId),"initialize_with_readme": "false"}
res = requests.post(url=targetHostUrl+'projects',data=prodata,headers={"PRIVATE-TOKEN":targetapiToken})
resdata = (json.loads(res.text));
return resdata['http_url_to_repo']
#获取所有项目
def getAllProject():
projectList = get(projectUrl)
for proj in projectList:
if proj['empty_repo']:
print('null project:',proj['path_with_namespace'])
else:
projInfo = {}
projInfo['id'] = proj['id']
projInfo['path_with_namespace'] = proj['path_with_namespace']
projInfo['empty_repo'] = proj['empty_repo']
projInfo['http_url_to_repo'] = proj['http_url_to_repo']
projInfo['default_branch'] = proj['default_branch']
projInfo['repo_branches'] = proj['_links']['repo_branches']
projectInfo.append(projInfo)
#写入拉取代码
projectCloneMirror.write("cd $workpath\n")
projectCloneMirror.write("git clone --mirror " + projInfo['http_url_to_repo'] + " " + projInfo['path_with_namespace'] +" \n")
#写入提交代码
projectCloneMirror.write("cd "+projInfo['path_with_namespace']+"\n")
projectCloneMirror.write("git push --mirror " + pushOnGroup(projInfo['path_with_namespace'],targetNamespaceId) +" \n")
return projectInfo
#
getAllProject()
#将所有项目信息写入文件
projectJson.write(json.dumps(projectInfo))
projectJson.flush()
projectJson.close()
#直接调起系统命令
print(os.system('git --version'))
六、脚本扩展说明
- git.py脚本中最后提供了调用命令行的方法,如果不想生成shell脚本,可再向文件wirte的地方更换成执行命令即可
- update.sh脚本中,自动识别到.git和.svn但未限制最大深度。
附录
官网:https://about.gitlab.com/
官方api地址: https://docs.gitlab.com/ee/api/文章来源地址https://www.toymoban.com/news/detail-430144.html
- gitlab帮助文档: http://127.0.0.1/help (自行替换ip和端口)
- 【推荐使用】gitlab rest api帮助文档: http://127.0.0.1/help/api/README.md (自行替换ip和端口)
- 【不推荐使用】gitlab graphql api帮助文档: http://127.0.0.1/help/api/graphql/index.md (自行替换ip和端口)
到了这里,关于【python】通过gitlab v4版本api接口批量拉取代码、迁移的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!