Work

domain/API.py

@@ -7,9 +7,10 @@ class API:
 	
 	__DEFAULT_API_PATH = "/api/v1"
 	
-	def __init__(self):
+	def __init__(self, verify_ssl, ca_bundle):
 	
-		pass
+		self.__verify_ssl = verify_ssl
+		self.__ca_bundle = ca_bundle
 	
 	@staticmethod
 	def _make_api_base_url(hostname, port):
@@ -20,17 +21,23 @@ class API:
 		
 		return base
 	
-	@staticmethod
+	def get(self, hostname, port, token) -> gitea.Gitea:
-	def factory(hostname, port, token) -> gitea.Gitea:
 		
 		url = API._make_api_base_url(
 			hostname=hostname,
 			port=port
 		)
 		
+		ssl_verify_arg = True
+		if self.__verify_ssl is not None:
+			ssl_verify_arg = self.__verify_ssl
+		if self.__ca_bundle is not None:
+			ssl_verify_arg = self.__ca_bundle
+		
 		g = gitea.Gitea(
 			gitea_url=url,
-			token_text=token
+			token_text=token,
+			verify=ssl_verify_arg
 		)
 		
 		return g

domain/Migrator.py

@@ -18,14 +18,13 @@ class Migrator:
 		self,
 		source_host, source_port, source_token,
 		destination_host, destination_port, destination_token,
+		verify_ssl: bool = True, ca_bundle: str = None
 	):
 		
 		# 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
@@ -34,15 +33,24 @@ class Migrator:
 		self.__destination_port = destination_port
 		self.__destination_token = destination_token
 		
-		self.__source_api = API.factory(
+		self.__verify_ssl = verify_ssl
+		self.__ca_bundle = ca_bundle
+		
+		api = API(
+			verify_ssl=self.__verify_ssl,
+			ca_bundle=self.__ca_bundle,
+		)
+		
+		self.__source_api = api.get(
 			hostname=self.__source_host,
 			port=self.__source_port,
-			token=self.__source_token
+			token=self.__source_token,
+			
 		)
-		self.__destination_api = API.factory(
+		self.__destination_api = api.get(
 			hostname=self.__destination_host,
 			port=self.__destination_port,
-			token=self.__destination_token
+			token=self.__destination_token,
 		)
 	
 	def _init_logger(self):
@@ -111,10 +119,7 @@ class Migrator:
 		
 		return repo_name
 	
-	def set_verify_ssl(self, b: bool):
+	"""
-		
-		self.__verify_ssl = b
-	
 	def set_ca_bundle(self, bundle_path: str):
 		
 		self.__logger.info("Setting certificate bundle path")
@@ -123,6 +128,10 @@ class Migrator:
 		self.__logger.info(f"Old path: {certifi.where()}")
 		certifi.core._CACERT_PATH = bundle_path
 		self.__logger.info(f"New path: {certifi.where()}")
+		
+		# TODO: JUST TESTING
+		self.__verify_ssl = bundle_path
+	"""
 	
 	def migrate_entire_org(
 			self,
@@ -165,19 +174,21 @@ class Migrator:
 		for repo in source_repos:
 			repo: gitea.Repository
 			self.__logger.info(f"- {repo.get_full_name()}")
-		assert False, "Poop"
 		
 		repos_migrate = []
 		repos_ignore = []
 		go_right_now = False
 		for repo in source_repos:
 			
-			repo: giteapy.Repository
+			repo: gitea.Repository
 			
 			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 = 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"
@@ -205,28 +216,32 @@ class Migrator:
 			if go_right_now:
 				break
 		
-		#
+		# Announce repo destination names
 		self.__logger.info("")
 		if len(repos_migrate):
 			self.__logger.info("Repos to migrate:")
 			for repo in repos_migrate:
-				repo: giteapy.Repository
+				repo: gitea.Repository
-				destination_name = self._make_destination_repo_name(pattern=destination_repo_name, repo=repo)
+				destination_name = self._make_destination_repo_name(
+					pattern=destination_repo_name, repo=repo
+				)
 				self.__logger.info(
 					f"#{repo.id} \"{repo.name}\"\n> \"{destination_name}\""
 				)
 		else:
 			self.__logger.info("No repos marked to migrate")
 		
+		# Announce manually ignored repos
 		self.__logger.info("")
 		if len(repos_ignore):
 			self.__logger.info("Repos to ignore:")
 			for repo in repos_ignore:
-				repo: giteapy.Repository
+				repo: gitea.Repository
-				self.__logger.info(f"#{repo.id} \"{repo.name}\"")
+				self.__logger.info(f"#{repo.id} \"{repo.get_full_name()}\"")
 		else:
 			self.__logger.info("No repos marked to ignore")
 		
+		# Migrate
 		if len(repos_migrate):
 			
 			confirmation = input("Do you confirm the above selections? Enter MIGRATE ==> ")
@@ -242,7 +257,8 @@ class Migrator:
 					repos=repos_migrate
 				)
 				self.__logger.info(
-					f"{len(source_repos_successful)} of {len(repos_migrate)} repos successfully migrated."
+					f"{len(source_repos_successful)} of {len(repos_migrate)}"
+					" repos successfully migrated."
 				)
 				if len(source_repos_failed) > 0:
 					self.__logger.error(f"Failed to migrate {len(source_repos_failed)} repos:")
@@ -256,7 +272,10 @@ class Migrator:
 							f"Failed to migrate repo: {repo.name}\n> {exception}"
 						)
 				
-				self._delete_migrated_repos(source_org_name=source_org, repos=source_repos_successful)
+				self._delete_migrated_repos(
+					source_org_name=source_org,
+					repos=source_repos_successful
+				)
 				
 			else:
 				self.__logger.info("Confirmation not received; Won't do anything.")
@@ -296,9 +315,7 @@ class Migrator:
 			repo_key = repo.get_full_name()
 			
 			topics_present = repo.get_topics()
-			print(topics_present)
+			repo_topics[repo_key] = topics_present
-			repo_topics[repo_key] = topics_present["topics"]
-			print(repo_topics[repo_key])
 			
 			if self._check_required_topics(
 				topics_present=repo_topics[repo_key],
@@ -334,9 +351,6 @@ class Migrator:
 	@staticmethod
 	def _check_required_topics(topics_present: list[str], topics_required: list[str]) -> bool:
 		
-		print("Required topics:", topics_required)
-		print("Present topics:", topics_present)
-		
 		for topic in topics_required:
 			if topic not in topics_present:
 				return False
@@ -352,44 +366,75 @@ class Migrator:
 			repos: list
 	):
 		
-		api_source, api_destination = self._get_org_apis()
+		# api_source, api_destination = self._get_org_apis()
+		# destination_org = api_destination.org_get(org=destination_org_name)
+		# destination_org: giteapy.Organization
 		
-		destination_org = api_destination.org_get(org=destination_org_name)
+		api_dest_org = gitea.Organization.request(
-		destination_org: giteapy.Organization
+			gitea=self.__destination_api,
-		self.__logger.info(f"Destination organization: {destination_org.full_name}")
+			name=destination_org_name
-		
-		api_source_repos = self._get_repo_api(
-			hostname=self.__source_host, port=self.__source_port,
-			token=self.__source_token
 		)
 		
+		self.__logger.info(f"Destination organization: {api_dest_org.full_name}")
+		
 		source_repos_successful = []
 		source_repos_failed = []
 		for source_repo in repos:
 			
-			source_repo: giteapy.Repository
+			source_repo: gitea.Repository
 			
-			this_destination_repo_name = destination_repo_name.replace("%N%", source_repo.name)
+			this_destination_repo_name = destination_repo_name.replace(
-			
+				"%N%",
-			self.__logger.info(f"Migrating: {source_repo.name} ==> {this_destination_repo_name}")
+				source_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
-			
-			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
 			)
+			
+			self.__logger.info(
+				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)
 			

main.py

@@ -138,13 +138,11 @@ def main():
 		source_token=args.source_token,
 		destination_host=args.destination_hostname,
 		destination_port=args.destination_port,
-		destination_token=args.destination_token
+		destination_token=args.destination_token,
+		verify_ssl=args.verify_ssl,
+		ca_bundle=args.ca_bundle
 	)
 	
-	mig.set_verify_ssl(args.verify_ssl)
-	if args.ca_bundle:
-		mig.set_ca_bundle(args.ca_bundle)
-	
 	mig.migrate_entire_org(
 		interactive=args.interactive,
 		source_org=args.source_org,