Migrate from gitlab.com to self-hosted gitlab

gitlab_migrate.py

"""
We used this script to migrate from gitlab.com to our self-hosted gitlab.
Author: Mohammad Javad Naderi

Before running the script, install `python-gitlab` package.

Important Note:
In order to keep the authors of merge requests and comments, before running this script make
sure all users have enabled "Public Email" in their gitlab.com profile and create account for
all users in self-hosted Gitlab (with the same e-mail address as their gitlab.com public email).
"""

import json
import logging
import os
import time

import gitlab

LOG_FORMAT = "%(asctime)s [%(levelname)s]\t %(message)s"
logging.basicConfig(format=LOG_FORMAT)
logging.getLogger().setLevel(logging.INFO)


class GitlabMigrate:
    def __init__(self, *, src_url, src_private_token, dst_url, dst_private_token):
        self.gl_src = gitlab.Gitlab(
            url=src_url, private_token=src_private_token, retry_transient_errors=True
        )
        self.gl_dst = gitlab.Gitlab(
            url=dst_url, private_token=dst_private_token, retry_transient_errors=True
        )

        # make sure urls and auth tokens are valid
        self.gl_src.auth()
        self.gl_dst.auth()

        self.users_dst = {
            user.email: user
            for user in [
                *self.gl_dst.users.list(page=1),
                *self.gl_dst.users.list(page=2),
                *self.gl_dst.users.list(page=3),
                *self.gl_dst.users.list(page=4),
            ]
            if user.email
        }

    @staticmethod
    def unprotect_project(project):
        try:
            project.protectedbranches.delete("*")
            return True
        except gitlab.exceptions.GitlabDeleteError:
            # already unprotected
            return False

    @staticmethod
    def protect_project(project):
        try:
            project.protectedbranches.create(
                {
                    "name": "*",
                    "merge_access_level": gitlab.const.AccessLevel.NO_ACCESS,
                    "push_access_level": gitlab.const.AccessLevel.NO_ACCESS,
                }
            )
            return True
        except gitlab.exceptions.GitlabCreateError:
            # already protected
            return False

    def migrate_project(self, project_src, namespace_dst: str):
        logging.info(
            f"--------------------------------------- Migrate Project: {project_src.path_with_namespace}"
        )
        dst_path_with_namespace = f"{namespace_dst}/{project_src.path}"

        # check if destination project already exists
        try:
            self.gl_dst.projects.get(dst_path_with_namespace)
            logging.warning("destination project already exists. doing nothing...")
            return
        except gitlab.exceptions.GitlabGetError:
            pass

        # Protect the src project from push or merge before export
        protected = GitlabMigrate.protect_project(project_src)
        if protected:
            logging.info("source project protected")

        # Create the export
        export = project_src.exports.create()
        logging.info(f"created export job")

        # Wait for the 'finished' status
        export.refresh()
        sleep_time = 5
        while export.export_status != "finished":
            logging.info(
                f"export status: {export.export_status}. waiting for {sleep_time} seconds..."
            )
            time.sleep(sleep_time)
            export.refresh()
            sleep_time = min(sleep_time + 2, 30)
        logging.info("project exported")

        export_file_path = f"/tmp/gitlab_export_{project_src.id}.tgz"

        # Download the result
        logging.info(f"downloading export to {export_file_path}")
        success = False
        while not success:
            try:
                with open(export_file_path, "wb") as f:
                    export.download(streamed=True, action=f.write)
                success = True
            except Exception:
                pass
        logging.info(
            f"export downloaded to {export_file_path} ({os.path.getsize(export_file_path) // 1024} kB)"
        )

        # Import
        with open(export_file_path, "rb") as f:
            output = self.gl_dst.projects.import_project(
                f, path=project_src.path, name=project_src.name, namespace=namespace_dst
            )
        logging.info(f"created import job: {output['id']}")

        # track the import status
        project_dst = self.gl_dst.projects.get(output["id"])
        project_import = project_dst.imports.get()
        sleep_time = 5
        while project_import.import_status != "finished":
            logging.info(
                f"import status: {project_import.import_status}. waiting for {sleep_time} seconds..."
            )
            time.sleep(sleep_time)
            project_import.refresh()
            sleep_time = min(sleep_time + 2, 30)
        logging.info("project imported")

        # Unrotect the dst project
        if protected:
            GitlabMigrate.unprotect_project(project_dst)
            logging.info("destination project unprotected")

        # logging.info("removing direct members")
        # for member in project_dst.members.list(iterator=True):
        #     member.delete()
        # logging.info("direct members removed")

        logging.info("migrate project finished.")
        return project_dst

    def migrate_group(self, group_src, parent_group_dst):
        logging.info(
            f"==================================================== Migrate Group: {group_src.full_path} ========="
        )

        try:
            group_dst = self.gl_dst.groups.get(
                f"{parent_group_dst.full_path}/{group_src.path}"
            )
            logging.warning(f"group already exists: {group_dst.id}")
        except gitlab.exceptions.GitlabGetError:
            group_dst = self.gl_dst.groups.create(
                {
                    "name": group_src.name,
                    "path": group_src.path,
                    "parent_id": parent_group_dst.id,
                }
            )
            logging.info(f"created group: {group_dst.id}")

            logging.info(f"adding group members")
            for member_src in group_src.members.list(iterator=True):
                user_src = self.gl_src.users.get(member_src.id)
                if user_src.public_email and (
                    user_dst := self.users_dst.get(user_src.public_email)
                ):
                    try:
                        group_dst.members.create(
                            {
                                "user_id": user_dst.id,
                                "access_level": member_src.access_level,
                            }
                        )
                    except gitlab.exceptions.GitlabCreateError:
                        pass  # member already exists
            logging.info(f"group members added")

            # Uncomment following lines if you want to change group member roles to "guest" in source
            # logging.info(f"making group members guest in src")
            # for member_src in group_src.members.list(iterator=True):
            #     logging.info(
            #         f"make guest | group: {group_src.full_path} | user: {member_src.username} | access_level: {member_src.access_level}"
            #     )
            #     member_src.access_level = gitlab.const.AccessLevel.GUEST
            #     try:
            #         member_src.save()
            #     except gitlab.exceptions.GitlabUpdateError:
            #         pass
            # logging.info(f"finished making group members guest in src")

        project_ids = set(prj.id for prj in group_src.projects.list(iterator=True))
        shared_project_ids = set(
            prj.id for prj in group_src.shared_projects.list(iterator=True)
        )
        direct_project_ids = project_ids - shared_project_ids
        for prj_id in direct_project_ids:
            project_src = self.gl_src.projects.get(prj_id)
            project_dst = self.migrate_project(
                project_src, namespace_dst=group_dst.full_path
            )

        for subgrp in group_src.subgroups.list(iterator=True):
            subgroup_src = self.gl_src.groups.get(subgrp.id)
            self.migrate_group(subgroup_src, group_dst)

        logging.info(
            f"#################################################### FINISH Migrate Group: {group_src.full_path} #########"
        )

    def run(self, *, src_group_id, dst_group_id):
        root_src = self.gl_src.groups.get(src_group_id)
        root_dst = self.gl_dst.groups.get(dst_group_id)
        self.migrate_group(root_src, root_dst)


if __name__ == "__main__":
    migrate = GitlabMigrate(
        src_url="https://gitlab.com",
        src_private_token="<PRIVATE-TOKEN>",
        dst_url="http://localhost:8123",
        dst_private_token="<PRIVATE-TOKEN>",
    )
    migrate.run(src_group_id=1234567, dst_group_id=1234)