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

Started refactoring to py-gitea

Browse Source
This commit is contained in:
mike
2024-06-17 23:45:40 -07:00
parent 16ab8a2ffd
commit 17a6422b1b
5 changed files with 245 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
domain/API.py Normal file
View File

@ -0,0 +1,36 @@
import gitea
class API:
__DEFAULT_API_PATH = "/api/v1"
def __init__(self):
pass
@staticmethod
def _make_api_base_url(hostname, port):
base = f"https://{hostname}"
if port is not None:
base += f":{port}"
return base
@staticmethod
def factory(hostname, port, token) -> gitea.Gitea:
url = API._make_api_base_url(
hostname=hostname,
port=port
)
g = gitea.Gitea(
gitea_url=url,
token_text=token
)
return g

136
domain/Migrator.py
View File

@ -1,6 +1,9 @@
import giteapy
from domain.API import API
import gitea
import logging
import sys
@ -9,7 +12,6 @@ import certifi
class Migrator:
__DEFAULT_API_PATH = "/api/v1"
__REPO_ORIGINAL_NAME_TOKEN = "%N%"
def __init__(
@ -31,6 +33,17 @@ class Migrator:
self.__destination_host = destination_host
self.__destination_port = destination_port
self.__destination_token = destination_token
self.__source_api = API.factory(
hostname=self.__source_host,
port=self.__source_port,
token=self.__source_token
)
self.__destination_api = API.factory(
hostname=self.__destination_host,
port=self.__destination_port,
token=self.__destination_token
)
def _init_logger(self):
@ -41,6 +54,7 @@ class Migrator:
self.__logger = logger
"""
def _get_user_api(self, hostname, port, token) -> giteapy.UserApi:
conf = giteapy.Configuration()
@ -50,7 +64,9 @@ class Migrator:
api = giteapy.UserApi(giteapy.ApiClient(conf))
return api
"""
"""
def _get_repo_api(self, hostname, port, token) -> giteapy.RepositoryApi:
conf = giteapy.Configuration()
@ -60,7 +76,9 @@ class Migrator:
api = giteapy.RepositoryApi(giteapy.ApiClient(conf))
return api
"""
"""
def _get_org_apis(self) -> (giteapy.OrganizationApi, giteapy.OrganizationApi):
api_source = self._get_org_api(
@ -73,7 +91,9 @@ class Migrator:
)
return api_source, api_destination
"""
"""
def _get_org_api(self, hostname, port, token) -> giteapy.OrganizationApi:
conf = giteapy.Configuration()
@ -83,17 +103,9 @@ class Migrator:
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):
def _make_destination_repo_name(self, pattern: str, repo: gitea.Repository):
repo_name = pattern.replace(self.__REPO_ORIGINAL_NAME_TOKEN, repo.name)
@ -126,24 +138,34 @@ class Migrator:
assert do_destination_copy_topics is not None, "Destination directive to copy source topics should be specified"
# api_source, api_destination = self._get_org_apis()
api_source: giteapy.OrganizationApi
api_destination: giteapy.OrganizationApi
# api_source: giteapy.OrganizationApi
# api_destination: giteapy.OrganizationApi
# Tattle on certify
self.__logger.info(f"Certifi is currently using CA bundle: {certifi.where()}")
# Grab all org repos
source_repos = self._fetch_all_org_repos(org=source_org)
source_repos = self._fetch_all_org_repos(
org_name=source_org
)
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}")
repo: gitea.Repository
self.__logger.info(f"- {repo.get_full_name()}")
print()
# Filter
source_repos = self._filter_repos_for_required_topics(repos=source_repos, topics=source_topics)
source_repos = self._filter_repos_for_required_topics(
repos=source_repos,
topics_required=source_topics
)
print()
self.__logger.info(f"Have {len(source_repos)} remaining repos after topic filtering:")
for repo in source_repos:
repo: gitea.Repository
self.__logger.info(f"- {repo.get_full_name()}")
assert False, "Poop"
repos_migrate = []
repos_ignore = []
@ -239,57 +261,58 @@ class Migrator:
else:
self.__logger.info("Confirmation not received; Won't do anything.")
def _fetch_all_org_repos(self, org: str):
def _fetch_all_org_repos(self, org_name: str):
api_source, api_destination = self._get_org_apis()
api_source: giteapy.OrganizationApi
org = gitea.Organization.request(
gitea=self.__source_api,
name=org_name
)
source_repos = []
# Grabs all pages automatically
repos = org.get_repositories()
page = 0
while True:
page += 1 # Starts at 1 for some reason
source_repos_page = api_source.org_list_repos(org, page=page, limit=25)
if len(source_repos_page) == 0:
break
source_repos.extend(source_repos_page)
return source_repos
return repos
def _filter_repos_for_required_topics(
self,
repos: list[giteapy.Repository],
topics: list[str]
repos: list[gitea.Repository],
topics_required: list[str]
) -> list[giteapy.Repository]:
) -> list[gitea.Repository]:
self.__logger.info(f"Filtering source repos for required topics: {topics}")
self.__logger.info(
f"Filtering source repos for required topics: {topics_required}"
)
repos_keep = []
repos_reject = []
repo_topics = {}
api_source_repos = self._get_repo_api(
hostname=self.__source_host, port=self.__source_port,
token=self.__source_token
)
for repo in repos:
repo_topics[repo.id] = api_source_repos.repo_list_topics(owner=repo.owner.login, repo=repo.name)
repo_topics[repo.id] = repo_topics[repo.id].topics
repo: gitea.Repository
if self._check_required_topics(topics_present=repo_topics[repo.id], topics_required=topics):
repo_key = repo.get_full_name()
topics_present = repo.get_topics()
print(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],
topics_required=topics_required
):
repos_keep.append(repo)
else:
repos_reject.append(repo)
self.__logger.info("")
self.__logger.info(f"\nKeeping {len(repos_keep)} repos because they contain all required topics ({topics}):")
self.__logger.info(
f"\nKeeping {len(repos_keep)} repos"
f" because they contain all required topics ({topics_required}):"
)
if len(repos_keep) > 0:
for repo in repos_keep:
self.__logger.info(f"> {repo.full_name}")
@ -297,10 +320,12 @@ class Migrator:
self.__logger.info("> None")
self.__logger.info("")
self.__logger.info(f"Rejecting {len(repos_reject)} repos because they don't contain all required topics:")
self.__logger.info(
f"Rejecting {len(repos_reject)} repos because they don't contain all required topics:"
)
if len(repos_reject) > 0:
for repo in repos_reject:
self.__logger.info(f"> {repo.full_name} ({repo_topics[repo.id]})")
self.__logger.info(f"> {repo.full_name}")
else:
self.__logger.info("> None")
@ -309,6 +334,9 @@ 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
@ -358,9 +386,9 @@ class Migrator:
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"
#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)
@ -416,7 +444,7 @@ class Migrator:
return source_repos_successful, source_repos_failed
def _delete_migrated_repos(self, source_org_name: str, repos: list[giteapy.Repository]):
def _delete_migrated_repos(self, source_org_name: str, repos: list[gitea.Repository]):
if len(repos) == 0:
self.__logger.warning(f"Cannot delete any migrated repos because none were successful!")