misc-utilities/gitea-repo-migrator
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Seems to be working.

Browse Source
This commit is contained in:
Mike 2023-01-18 00:47:08 -08:00
parent 95b494e547
commit 74aa794571
1 changed files with 113 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
domain/Migrator.py
View File

@ -39,7 +39,17 @@ class Migrator:
self.__logger = logger
def _get_repo_api(self, hostname, port, token):
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
@ -49,7 +59,7 @@ class Migrator:
return api
def _get_org_apis(self):
def _get_org_apis(self) -> (giteapy.OrganizationApi, giteapy.OrganizationApi):
api_source = self._get_org_api(
hostname=self.__source_host, port=self.__source_port,
@ -62,7 +72,7 @@ class Migrator:
return api_source, api_destination
def _get_org_api(self, hostname, port, token):
def _get_org_api(self, hostname, port, token) -> giteapy.OrganizationApi:
conf = giteapy.Configuration()
conf.api_key['access_token'] = token
@ -163,15 +173,22 @@ class Migrator:
self.__logger.info("No repos marked to ignore")
if len(repos_migrate):
confirmation = input("Do you confirm the above selections? Enter CONFIRM ==> ")
if confirmation == "CONFIRM":
confirmation = input("Do you confirm the above selections? Enter MIGRATE ==> ")
if confirmation == "MIGRATE":
self.__logger.info("Confirmation received; Processing ... ")
self._migrate_repos(
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.")
@ -190,35 +207,113 @@ class Migrator:
self.__logger.info(f"Destination organization: {destination_org.full_name}")
for repo in repos:
source_repos_successful = []
for source_repo in repos:
repo: giteapy.Repository
source_repo: giteapy.Repository
this_destination_repo_name = destination_repo_name.replace("%N%", repo.name)
this_destination_repo_name = destination_repo_name.replace("%N%", source_repo.name)
migrate_body = giteapy.MigrateRepoForm(
mirror=False,
clone_addr=repo.clone_url,
clone_addr=source_repo.clone_url,
uid=destination_org.id,
private=repo.private,
private=source_repo.private,
repo_name=this_destination_repo_name,
description=repo.description,
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.info("Migrate body:")
self.__logger.info(migrate_body)
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,
)
migration_result = destination_api.repo_migrate(body=migrate_body)
self.__logger.info(f"Migration result: {migration_result}")
repo_new = destination_api.repo_migrate(body=migrate_body)
self.__logger.debug(f"Migration result: {repo_new}")
repo_new: giteapy.Repository
for t in destination_topics:
self.__logger.info(f"Want to append topic: {t}")
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
)