351 lines
10 KiB
Python
351 lines
10 KiB
Python
|
|
|
|
# import giteapy
|
|
import giteapy.giteapy
|
|
import logging
|
|
import sys
|
|
|
|
|
|
class Migrator:
|
|
|
|
__DEFAULT_API_PATH = "/api/v1"
|
|
__REPO_ORIGINAL_NAME_TOKEN = "%N%"
|
|
|
|
def __init__(
|
|
self,
|
|
source_host, source_port, source_token,
|
|
destination_host, destination_port, destination_token,
|
|
):
|
|
|
|
# noinspection PyTypeChecker
|
|
self.__logger: logging.Logger = None
|
|
self._init_logger()
|
|
|
|
self.__verify_ssl = True
|
|
|
|
self.__source_host = source_host
|
|
self.__source_port = source_port
|
|
self.__source_token = source_token
|
|
|
|
self.__destination_host = destination_host
|
|
self.__destination_port = destination_port
|
|
self.__destination_token = destination_token
|
|
|
|
def _init_logger(self):
|
|
|
|
logger = logging.Logger(name=f"{type(self).__name__}", level=logging.INFO)
|
|
|
|
stdout_handler = logging.StreamHandler(stream=sys.stdout)
|
|
logger.addHandler(stdout_handler)
|
|
|
|
self.__logger = logger
|
|
|
|
def _get_user_api(self, hostname, port, token) -> giteapy.UserApi:
|
|
|
|
conf = giteapy.Configuration()
|
|
conf.api_key['access_token'] = token
|
|
conf.host = self._make_api_base(hostname=hostname, port=port)
|
|
conf.verify_ssl = self.__verify_ssl
|
|
api = giteapy.UserApi(giteapy.ApiClient(conf))
|
|
|
|
return api
|
|
|
|
def _get_repo_api(self, hostname, port, token) -> giteapy.RepositoryApi:
|
|
|
|
conf = giteapy.Configuration()
|
|
conf.api_key['access_token'] = token
|
|
conf.host = self._make_api_base(hostname=hostname, port=port)
|
|
conf.verify_ssl = self.__verify_ssl
|
|
api = giteapy.RepositoryApi(giteapy.ApiClient(conf))
|
|
|
|
return api
|
|
|
|
def _get_org_apis(self) -> (giteapy.OrganizationApi, giteapy.OrganizationApi):
|
|
|
|
api_source = self._get_org_api(
|
|
hostname=self.__source_host, port=self.__source_port,
|
|
token=self.__source_token
|
|
)
|
|
api_destination = self._get_org_api(
|
|
hostname=self.__destination_host, port=self.__destination_port,
|
|
token=self.__destination_token
|
|
)
|
|
|
|
return api_source, api_destination
|
|
|
|
def _get_org_api(self, hostname, port, token) -> giteapy.OrganizationApi:
|
|
|
|
conf = giteapy.Configuration()
|
|
conf.api_key['access_token'] = token
|
|
conf.host = self._make_api_base(hostname=hostname, port=port)
|
|
conf.verify_ssl = self.__verify_ssl
|
|
api = giteapy.OrganizationApi(giteapy.ApiClient(conf))
|
|
|
|
return api
|
|
|
|
def _make_api_base(self, hostname, port):
|
|
|
|
base = f"https://{hostname}"
|
|
if port is not None:
|
|
base += f":{port}"
|
|
base += self.__DEFAULT_API_PATH
|
|
|
|
return base
|
|
|
|
def _make_destination_repo_name(self, pattern: str, repo: giteapy.Repository):
|
|
|
|
repo_name = pattern.replace(self.__REPO_ORIGINAL_NAME_TOKEN, repo.name)
|
|
|
|
return repo_name
|
|
|
|
def set_verify_ssl(self, b: bool):
|
|
|
|
self.__verify_ssl = b
|
|
|
|
def migrate_entire_org(
|
|
self,
|
|
interactive: bool = True,
|
|
source_org: str = None, source_topics: list[str] = None,
|
|
destination_org: str = None, destination_repo_name: str = None, destination_topics: list = None
|
|
):
|
|
assert source_org is not None, "Source org must be specified"
|
|
assert destination_org is not None, "Destination org must be specified"
|
|
assert destination_repo_name is not None, "Destination repo name must be specified"
|
|
assert destination_topics is not None, "Destination topics must be specified"
|
|
|
|
api_source, api_destination = self._get_org_apis()
|
|
api_source: giteapy.OrganizationApi
|
|
api_destination: giteapy.OrganizationApi
|
|
|
|
api_source_repos = self._get_repo_api(hostname=self.__source_host, port=self.__source_port, token=self.__source_token)
|
|
|
|
source_repos = api_source.org_list_repos(source_org, page=0, limit=1000000)
|
|
self.__logger.info(f"Found {len(source_repos)} repos on source:")
|
|
for repo in source_repos:
|
|
repo: giteapy.Repository
|
|
self.__logger.info(f"- #{repo.id} {repo.full_name}")
|
|
|
|
print()
|
|
|
|
repos_migrate = []
|
|
repos_ignore = []
|
|
go_right_now = False
|
|
for repo in source_repos:
|
|
|
|
repo: giteapy.Repository
|
|
|
|
repo_topics = api_source_repos.repo_list_topics(owner=repo.owner.login, repo=repo.name)
|
|
repo_topics = repo_topics.topics
|
|
self.__logger.error(f"Repo topics: {repo_topics}")
|
|
|
|
if self._check_source_repo_topics(repo=repo, topics=source_topics) is False:
|
|
repos_ignore.append(repo)
|
|
self.__logger.info(
|
|
f"Ignoring repo because it doesn't have all required tags."
|
|
f"\n> Repo: {repo.full_name}"
|
|
f"\n> Has: {repo_topics}"
|
|
f"\n> Needs: {source_topics}"
|
|
)
|
|
continue
|
|
|
|
while True:
|
|
|
|
if interactive:
|
|
response = input(f"Migrate repo #{repo.id} \"{repo.full_name}\" ? (Y)es, (N)o, (G)o right now, (Q)uit ==> ")
|
|
response = response.lower()
|
|
else:
|
|
response = "y"
|
|
|
|
valid_input = True
|
|
if response == "y":
|
|
repos_migrate.append(repo)
|
|
elif response == "n":
|
|
repos_ignore.append(repo)
|
|
elif response == "g":
|
|
self.__logger.info("Okay, done asking questions, migrating existing selections.")
|
|
go_right_now = True
|
|
elif response == "q":
|
|
go_right_now = True
|
|
repos_migrate.clear()
|
|
repos_ignore.clear()
|
|
self.__logger.info("Okay, quitting instead.")
|
|
else:
|
|
valid_input = False
|
|
self.__logger.warning(f"Invalid input: {response}")
|
|
|
|
if valid_input:
|
|
break
|
|
|
|
if go_right_now:
|
|
break
|
|
|
|
#
|
|
self.__logger.info("")
|
|
if len(repos_migrate):
|
|
self.__logger.info("Repos to migrate:")
|
|
for repo in repos_migrate:
|
|
repo: giteapy.Repository
|
|
destination_name = self._make_destination_repo_name(pattern=destination_repo_name, repo=repo)
|
|
self.__logger.info(f"#{repo.id} \"{repo.name}\" ==> \"{destination_name}\"")
|
|
else:
|
|
self.__logger.info("No repos marked to migrate")
|
|
|
|
self.__logger.info("")
|
|
if len(repos_ignore):
|
|
self.__logger.info("Repos to ignore:")
|
|
for repo in repos_ignore:
|
|
repo: giteapy.Repository
|
|
self.__logger.info(f"#{repo.id} \"{repo.name}\"")
|
|
else:
|
|
self.__logger.info("No repos marked to ignore")
|
|
|
|
if len(repos_migrate):
|
|
|
|
confirmation = input("Do you confirm the above selections? Enter MIGRATE ==> ")
|
|
|
|
if confirmation == "MIGRATE":
|
|
|
|
self.__logger.info("Confirmation received; Processing ... ")
|
|
source_repos_successful = self._migrate_repos(
|
|
destination_org_name=destination_org,
|
|
destination_repo_name=destination_repo_name,
|
|
destination_topics=destination_topics,
|
|
repos=repos_migrate
|
|
)
|
|
self.__logger.info(f"{len(source_repos_successful)} of {len(repos_migrate)} repos successfully migrated.")
|
|
|
|
self._delete_migrated_repos(source_org_name=source_org, repos=source_repos_successful)
|
|
|
|
else:
|
|
self.__logger.info("Confirmation not received; Won't do anything.")
|
|
|
|
def _check_source_repo_topics(self, repo: giteapy.Repository, topics: list[str]) -> bool:
|
|
|
|
return False
|
|
|
|
def _migrate_repos(
|
|
self,
|
|
destination_org_name: str,
|
|
destination_repo_name: str,
|
|
destination_topics: list,
|
|
repos: list
|
|
):
|
|
|
|
api_source, api_destination = self._get_org_apis()
|
|
|
|
destination_org = api_destination.org_get(org=destination_org_name)
|
|
destination_org: giteapy.Organization
|
|
|
|
self.__logger.info(f"Destination organization: {destination_org.full_name}")
|
|
|
|
source_repos_successful = []
|
|
for source_repo in repos:
|
|
|
|
source_repo: giteapy.Repository
|
|
|
|
this_destination_repo_name = destination_repo_name.replace("%N%", source_repo.name)
|
|
|
|
migrate_body = giteapy.MigrateRepoForm(
|
|
mirror=False,
|
|
clone_addr=source_repo.clone_url,
|
|
uid=destination_org.id,
|
|
private=source_repo.private,
|
|
repo_name=this_destination_repo_name,
|
|
description=source_repo.description,
|
|
labels=True, issues=True, pull_requests=True, releases=True, milestones=True, wiki=True
|
|
)
|
|
# TODO: These three lines represent feature request to giteapy authors
|
|
migrate_body.auth_token = self.__source_token
|
|
migrate_body.swagger_types["auth_token"] = "str"
|
|
migrate_body.attribute_map["auth_token"] = "auth_token"
|
|
|
|
self.__logger.debug("Migrate body:")
|
|
self.__logger.debug(migrate_body)
|
|
|
|
destination_api = self._get_repo_api(
|
|
hostname=self.__destination_host,
|
|
port=self.__destination_port,
|
|
token=self.__destination_token,
|
|
)
|
|
repo_new = destination_api.repo_migrate(body=migrate_body)
|
|
self.__logger.debug(f"Migration result: {repo_new}")
|
|
repo_new: giteapy.Repository
|
|
|
|
assert repo_new.name == this_destination_repo_name,\
|
|
"New repository didn't end up with the correct name. Failure?"
|
|
|
|
for topic in destination_topics:
|
|
self.__logger.debug(f"Appending topic to new repo: {topic}")
|
|
destination_api.repo_add_topc(
|
|
owner=destination_org.username,
|
|
repo=repo_new.name,
|
|
topic=topic,
|
|
)
|
|
|
|
source_repos_successful.append(source_repo)
|
|
|
|
return source_repos_successful
|
|
|
|
def _delete_migrated_repos(self, source_org_name: str, repos: list[giteapy.Repository]):
|
|
|
|
repo_api = self._get_repo_api(
|
|
hostname=self.__source_host,
|
|
port=self.__source_port,
|
|
token=self.__source_token
|
|
)
|
|
repo_api: giteapy.RepositoryApi
|
|
|
|
self.__logger.info("")
|
|
self.__logger.info("Can now delete the following successfully migrated repos:")
|
|
for r in repos:
|
|
self.__logger.info(f"> #{r.id} \"{r.full_name}\" ==> {r.clone_url}")
|
|
|
|
response = input("Would you like to delete the successfully migrated repos? Type DELETE ==> ")
|
|
|
|
if response != "DELETE":
|
|
self.__logger.info("Okay, won't delete migrated repos.")
|
|
return
|
|
|
|
do_quit = False
|
|
do_delete_all = False
|
|
for repo in repos:
|
|
|
|
self.__logger.info(f"Next repo to delete: #{repo.id} \"{repo.full_name}\"")
|
|
do_delete = True if do_delete_all else False
|
|
|
|
if do_delete is False:
|
|
|
|
response = input("Delete this repo? (Y)es, (N), (A)ll, (Q)uit ==> ")
|
|
response = response.lower()
|
|
|
|
valid_response = False
|
|
|
|
while valid_response is False:
|
|
|
|
valid_response = True
|
|
if response == "y":
|
|
self.__logger.info(f"Okay, deleting {repo.name} from source")
|
|
do_delete = True
|
|
elif response == "n":
|
|
self.__logger.info(f"Okay, won't delete {repo.name}")
|
|
elif response == "a":
|
|
self.__logger.info(f"Okay, deleting ALL remaining repos")
|
|
do_delete_all = True
|
|
elif response == "q":
|
|
do_quit = True
|
|
else:
|
|
valid_response = False
|
|
self.__logger.warning(f"Invalid response: {response}")
|
|
|
|
if do_quit:
|
|
break
|
|
|
|
if do_delete or do_delete_all:
|
|
|
|
self.__logger.info(f"Deleting repo: {repo.full_name}")
|
|
|
|
repo_api.repo_delete(
|
|
owner=source_org_name,
|
|
repo=repo.name
|
|
)
|