From ec81cab79d202fd1d7942de69af61a57b43ff451 Mon Sep 17 00:00:00 2001 From: mike Date: Tue, 18 Jun 2024 02:04:43 -0700 Subject: [PATCH] Seems to be working now --- domain/Migrator.py | 136 +++++++++++++++++---------------------------- 1 file changed, 50 insertions(+), 86 deletions(-) diff --git a/domain/Migrator.py b/domain/Migrator.py index e6b7667..6f1acc0 100644 --- a/domain/Migrator.py +++ b/domain/Migrator.py @@ -392,98 +392,64 @@ class Migrator: f"Migrating: {source_repo.name} ==> {this_destination_repo_name}" ) - # source_repo_topics = api_source_repos.repo_list_topics( - # owner=source_repo.owner.login, repo=source_repo.name - # ) - # source_repo_topics = source_repo_topics.topics - topics = source_repo.get_topics() - - # 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" - - destination_repo = gitea.Repository.migrate_repo( - gitea=self.__destination_api, - service: str, - clone_addr: str, - repo_name: str, - description: str = "", - private: bool = False, - auth_token: str = None, - auth_username: str = None, - auth_password: str = None, - mirror: bool = False, - mirror_interval: str = None, - lfs: bool = False, - lfs_endpoint: str = "", - wiki: bool = False, - labels: bool = False, - issues: bool = False, - pull_requests: bool = False, - releases: bool = False, - milestones: bool = False, - repo_owner: str = None, - ) - - self.__logger.debug("Migrate body:") - self.__logger.debug(migrate_body) + source_repo_topics = source_repo.get_topics() try: - destination_api = self._get_repo_api( - hostname=self.__destination_host, - port=self.__destination_port, - token=self.__destination_token, + + repo_new = gitea.Repository.migrate_repo( + gitea=self.__destination_api, + service="gitea", # type of remote service + clone_addr=source_repo.clone_url, + repo_name=this_destination_repo_name, + description=source_repo.description, + private=source_repo.private, + auth_token=self.__source_token, + auth_username=None, + auth_password=None, + mirror=False, + mirror_interval=None, + # lfs=False, + # lfs_endpoint="", + wiki=True, + labels=True, + issues=True, + pull_requests=True, + releases=True, + milestones=True, + repo_owner=destination_org_name, ) - except giteapy.rest.ApiException as e: - self.__logger.error(f"Failed to generate destination API: {e}") - source_repos_failed.append( - (source_repo, e) - ) - continue - try: - repo_new = destination_api.repo_migrate(body=migrate_body) - except giteapy.rest.ApiException as e: - self.__logger.error(f"Failed to execute repo migration request via API: {e}") + except Exception as e: + self.__logger.error( + f"Failed to execute repo migration request:" + f"\n{e}" + ) source_repos_failed.append( (source_repo, e) ) continue self.__logger.debug(f"Migration result: {repo_new}") - repo_new: giteapy.Repository + repo_new: gitea.Repository - assert repo_new.name == this_destination_repo_name,\ + assert repo_new.name == this_destination_repo_name, \ "New repository didn't end up with the correct name. Failure?" # Copy source topics? if do_destination_copy_topics: + for topic in source_repo_topics: + self.__logger.debug(f"Appending source topic to new repo: {topic}") - destination_api.repo_add_topc( - owner=destination_org.username, - repo=repo_new.name, - topic=topic, - ) + + repo_new.add_topic(topic=topic) # Add specified topics 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, - ) + + repo_new.add_topic(topic=topic) source_repos_successful.append(source_repo) @@ -495,20 +461,19 @@ class Migrator: self.__logger.warning(f"Cannot delete any migrated repos because none were successful!") return - 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(f"Can now delete {len(repos)} successfully migrated repos:") + self.__logger.info(f"Can now delete repos from source org: {source_org_name}") + self.__logger.info(f"Will delete {len(repos)} successfully migrated repos:") for r in repos: + + r: gitea.Repository + 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 ==> ") - + # Ask the user to confirm deletion + 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 @@ -517,7 +482,9 @@ class Migrator: do_delete_all = False for repo in repos: - self.__logger.info(f"Next repo to delete: #{repo.id} \"{repo.full_name}\"") + repo: gitea.Repository + + self.__logger.info(f"Next repo to delete: \"{repo.full_name}\"") do_delete = True if do_delete_all else False if do_delete is False: @@ -551,7 +518,4 @@ class Migrator: self.__logger.info(f"Deleting repo: {repo.full_name}") - repo_api.repo_delete( - owner=source_org_name, - repo=repo.name - ) + repo.delete()