diff --git a/domain/Migrator.py b/domain/Migrator.py index 359b614..3941811 100644 --- a/domain/Migrator.py +++ b/domain/Migrator.py @@ -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 + )