当前位置：首页 > article >正文

gitlab初始化+API批量操作

article 2025/1/6 19:27:25

几年没接触gitlab了，新版本装完以后代码提交到默认的main分支，master不再是主分支
项目有几十个仓库，研发提交代码后仓库地址和之前的发生了变化
有几个点需要注意
1、修改全局默认分支
2、关闭分支保护
在这里插入图片描述

上面修改了全局配置不会影响已经创建好的项目，所以还需要批量修改已创建的项目
先修改Group的默认分支，不会影响已存在的项目在这里插入图片描述
修改gitlab全局的默认分支

这就引出了需求，要获取当前仓库下所有仓库地址，还要修改所有仓库的默认分支，从main修改为master

生成AccessToken

创建个人访问令牌

填写信息时，确保勾选以下权限（Scopes）：
api
read_api
read_repository
write_repository
admin_mode（如果是要批量修改默认分支，可能需要这个权限）
# 权限解析
api 和 read_api: 基本的 API 访问权限
read_repository: 读取仓库信息
write_repository: 修改仓库设置（包括默认分支）
admin_mode: 管理员级别的操作权限

需求1-获取仓库下所有项目地址

import requests
import sys
import time
from typing import List, Dict

class GitLabProjectFetcher:
    def __init__(self, base_url: str, private_token: str):
        """
        初始化 GitLab API 客户端
        
        Args:
            base_url: GitLab 实例的基础 URL
            private_token: GitLab 个人访问令牌
        """
        self.base_url = base_url.rstrip('/')
        self.headers = {'PRIVATE-TOKEN': private_token}
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def get_all_projects(self) -> List[Dict]:
        """获取所有项目"""
        projects = []
        page = 1
        while True:
            url = f"{self.base_url}/api/v4/projects"
            params = {
                'page': page,
                'per_page': 100,
                'order_by': 'path',
                'sort': 'asc'
            }
            
            try:
                response = self.session.get(url, params=params)
                response.raise_for_status()
                
                batch = response.json()
                if not batch:
                    break
                    
                projects.extend(batch)
                page += 1
                print(f"已获取 {len(projects)} 个项目...")
                
                time.sleep(0.5)  # 避免请求过快
                
            except requests.exceptions.RequestException as e:
                print(f"获取项目列表时发生错误: {str(e)}")
                sys.exit(1)
            
        return projects

def main():
    # 配置信息 - 根据实际环境修改
    GITLAB_URL = "http://gitlab地址"
    PRIVATE_TOKEN = "accesstoken"
    
    fetcher = GitLabProjectFetcher(GITLAB_URL, PRIVATE_TOKEN)
    
    try:
        print("正在获取项目列表...")
        projects = fetcher.get_all_projects()
        
        print("\n所有项目的 Git SSH Clone URL:")
        print("-" * 80)
        
        # 按照路径排序
        sorted_projects = sorted(projects, key=lambda x: x['path_with_namespace'])
        
        for project in sorted_projects:
            if 'ssh_url_to_repo' in project:
                print(project['ssh_url_to_repo'])
        
        print(f"\n总共找到 {len(projects)} 个项目")
        
    except Exception as e:
        print(f"发生错误: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()

需求2-修改所有仓库默认的分支为master

root@prd-ops-cicd-jenkins33:/data/ops# cat chage-branch.py 
import requests
import sys
import time
from typing import List, Dict

class GitLabProjectManager:
    def __init__(self, base_url: str, private_token: str):
        """
        初始化 GitLab API 客户端
        """
        self.base_url = base_url.rstrip('/')
        self.headers = {'PRIVATE-TOKEN': private_token}
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def get_all_projects(self) -> List[Dict]:
        """获取所有项目"""
        projects = []
        page = 1
        while True:
            url = f"{self.base_url}/api/v4/projects"
            params = {
                'page': page,
                'per_page': 100,
                'order_by': 'path',
                'sort': 'asc',
                'simple': False
            }
            
            try:
                print(f"正在获取第 {page} 页...")
                response = self.session.get(url, params=params)
                response.raise_for_status()
                
                batch = response.json()
                if not batch:
                    break
                    
                projects.extend(batch)
                page += 1
                time.sleep(0.5)
                
            except requests.exceptions.RequestException as e:
                print(f"获取项目列表时发生错误: {str(e)}")
                sys.exit(1)
            
        return projects

    def check_branch_exists(self, project_id: int, branch: str) -> bool:
        """检查分支是否存在"""
        url = f"{self.base_url}/api/v4/projects/{project_id}/repository/branches/{branch}"
        response = self.session.get(url)
        return response.status_code == 200

    def update_default_branch(self, project_id: int, project_name: str, new_branch: str) -> bool:
        """更新项目的默认分支"""
        # 检查目标分支是否存在
        if not self.check_branch_exists(project_id, new_branch):
            print(f"项目 {project_name} 中不存在 {new_branch} 分支，跳过")
            return False

        url = f"{self.base_url}/api/v4/projects/{project_id}"
        data = {'default_branch': new_branch}
        
        try:
            response = self.session.put(url, json=data)
            response.raise_for_status()
            return True
        except requests.exceptions.RequestException as e:
            print(f"更新失败: {str(e)}")
            return False

def main():
    # 配置信息
    GITLAB_URL = "http://gitlab地址"
    PRIVATE_TOKEN = "accesstoken"
    NEW_DEFAULT_BRANCH = "master"
    
    manager = GitLabProjectManager(GITLAB_URL, PRIVATE_TOKEN)
    
    try:
        # 获取所有项目
        print("正在获取项目列表...")
        projects = manager.get_all_projects()
        print(f"\n找到 {len(projects)} 个项目")
        
        # 更新默认分支
        success_count = 0
        fail_count = 0
        skipped_count = 0
        
        print("\n开始更新默认分支...")
        for project in projects:
            project_id = project['id']
            project_name = project['path_with_namespace']
            current_default = project.get('default_branch', '')
            
            print(f"\n处理项目: {project_name}")
            print(f"当前默认分支: {current_default}")
            
            if current_default == NEW_DEFAULT_BRANCH:
                print("默认分支已经是 master，跳过")
                skipped_count += 1
                continue
                
            if manager.update_default_branch(project_id, project_name, NEW_DEFAULT_BRANCH):
                success_count += 1
                print(f"✓ 成功将默认分支更新为 {NEW_DEFAULT_BRANCH}")
            else:
                fail_count += 1
                print(f"✗ 更新失败")
        
        # 打印总结
        print("\n更新完成！")
        print(f"成功更新: {success_count} 个项目")
        print(f"更新失败: {fail_count} 个项目")
        print(f"跳过项目: {skipped_count} 个项目")
        print(f"总项目数: {len(projects)} 个项目")
        
    except Exception as e:
        print(f"发生错误: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()