diff --git a/src/forgesync/cli.py b/src/forgesync/cli.py index d6059e7..5552dfc 100644 --- a/src/forgesync/cli.py +++ b/src/forgesync/cli.py @@ -2,7 +2,6 @@ Automatically synchronize all your Forgejo repositories to GitHub as well as any Forgejo instance. """ -from enum import StrEnum from logging import Formatter, Logger, StreamHandler from os import environ, PathLike from sys import stderr @@ -10,22 +9,17 @@ from tap import Tap from xdg_base_dirs import xdg_config_home from pyforgejo import PyforgejoApi, Repository as ForgejoRepository -from forgesync.sync import SyncError, SyncedRepository +from .sync import SyncError, SyncedRepository, Destination from .github import GithubSyncer from .forgejo import ForgejoSyncer -from .mirror import MirrorError, PushMirrorer +from .mirror import MirrorError, PushMirrorConfig, PushMirrorer from re import compile -class ToType(StrEnum): - GITHUB = "github" - FORGEJO = "forgejo" - - class ArgumentParser(Tap): from_instance: str "base URL of the source instance" - to: ToType + to: Destination "what kind of destination to sync to, e.g. 'forgejo' or 'github'" to_instance: str "base URL of the destination instance" @@ -99,22 +93,37 @@ def main() -> None: logger = make_logger(name="forgesync", level=args.log) + from_client = PyforgejoApi(base_url=args.from_instance, api_key=from_token) + + push_mirror_config = PushMirrorConfig( + interval=args.mirror_interval, + remirror=args.remirror, + immediate=args.immediate, + sync_on_commit=args.sync_on_commit, + ) + + push_mirrorer = PushMirrorer( + client=from_client, + config=push_mirror_config, + mirror_token=mirror_token, + logger=logger, + ) + match args.to: - case ToType.GITHUB: + case Destination.GITHUB: syncer = GithubSyncer( instance=args.to_instance, token=to_token, + push_mirrorer=push_mirrorer, logger=logger, ) - case ToType.FORGEJO: + case Destination.FORGEJO: syncer = ForgejoSyncer( instance=args.to_instance, token=to_token, logger=logger, ) - from_client = PyforgejoApi(base_url=args.from_instance, api_key=from_token) - from_user = from_client.user.get_current() if from_user.login is None: logger.fatal("Could not get username from Forgejo") @@ -148,19 +157,10 @@ def main() -> None: synced_repos.append(synced_repo) - push_mirrorer = PushMirrorer( - client=from_client, - logger=logger, - ) - try: _ = push_mirrorer.mirror_repos( synced_repos=synced_repos, - interval=args.mirror_interval, - remirror=args.remirror, - immediate=args.immediate, - sync_on_commit=args.sync_on_commit, - mirror_token=mirror_token, + config=push_mirror_config, ) except MirrorError as error: logger.fatal("Mirroring failed: %s", error) diff --git a/src/forgesync/forgejo.py b/src/forgesync/forgejo.py index 65e1056..051f77f 100644 --- a/src/forgesync/forgejo.py +++ b/src/forgesync/forgejo.py @@ -1,7 +1,8 @@ from logging import Logger from typing import Self, override from pyforgejo import PyforgejoApi, Repository as ForgejoRepository, User as ForgejoUser -from .sync import SyncError, SyncedRepository, Syncer + +from .sync import SyncError, SyncedRepository, Syncer, Destination class ForgejoSyncer(Syncer): @@ -10,7 +11,12 @@ class ForgejoSyncer(Syncer): repos: dict[str, ForgejoRepository] logger: Logger - def __init__(self: Self, instance: str, token: str, logger: Logger) -> None: + def __init__( + self: Self, + instance: str, + token: str, + logger: Logger, + ) -> None: self.client = PyforgejoApi(base_url=instance, api_key=token) self.user = self.client.user.get_current() @@ -28,7 +34,9 @@ class ForgejoSyncer(Syncer): @override def sync( - self: Self, from_repo: ForgejoRepository, description: str + self: Self, + from_repo: ForgejoRepository, + description: str, ) -> SyncedRepository: assert from_repo.name is not None self.logger.info("Synchronizing %s", from_repo.name) @@ -44,7 +52,8 @@ class ForgejoSyncer(Syncer): description=description, private=from_repo.private, ) - self.logger.info(f"Created new Forgejo repository %s", new_repo.full_name) + + self.logger.info("Created new Forgejo repository %s", new_repo.full_name) edited_repo = self.client.repository.repo_edit( owner=self.user.login, @@ -87,4 +96,6 @@ class ForgejoSyncer(Syncer): orig_owner=from_repo.owner.login, name=edited_repo.name, clone_url=edited_repo.clone_url, + destination=Destination.FORGEJO, + needs_mirror=True, ) diff --git a/src/forgesync/github.py b/src/forgesync/github.py index de0e3cc..24a5762 100644 --- a/src/forgesync/github.py +++ b/src/forgesync/github.py @@ -1,20 +1,30 @@ from logging import Logger from typing import Self, override from github.AuthenticatedUser import AuthenticatedUser +from github.GithubException import GithubException from github.GithubObject import NotSet from github.Repository import Repository as GithubRepository from github import Github, Auth as GithubAuth from pyforgejo import Repository as ForgejoRepository -from .sync import SyncError, SyncedRepository, Syncer + +from .mirror import PushMirrorConfig, PushMirrorer +from .sync import SyncError, SyncedRepository, Syncer, Destination class GithubSyncer(Syncer): client: Github user: AuthenticatedUser repos: dict[str, GithubRepository] + push_mirrorer: PushMirrorer logger: Logger - def __init__(self: Self, instance: str, token: str, logger: Logger) -> None: + def __init__( + self: Self, + instance: str, + token: str, + push_mirrorer: PushMirrorer, + logger: Logger, + ) -> None: auth = GithubAuth.Token(token) self.client = Github( @@ -33,17 +43,36 @@ class GithubSyncer(Syncer): for repo in self.user.get_repos(): self.repos[repo.name] = repo + self.push_mirrorer = push_mirrorer + self.logger = logger @override def sync( - self: Self, from_repo: ForgejoRepository, description: str + self: Self, + from_repo: ForgejoRepository, + description: str, ) -> SyncedRepository: if from_repo.name is None: raise SyncError("could not get Forgejo repository name") self.logger.info("Synchronizing %s", from_repo.name) + needs_mirror = True + + def make_synced(repo: GithubRepository) -> SyncedRepository: + if from_repo.owner is None or from_repo.owner.login is None: + raise SyncError("received malformed repository") + + return SyncedRepository( + new_owner=repo.owner.login, + orig_owner=from_repo.owner.login, + name=repo.name, + clone_url=repo.clone_url, + destination=Destination.GITHUB, + needs_mirror=True, + ) + if from_repo.name in self.repos: repo = self.repos[from_repo.name] else: @@ -59,8 +88,38 @@ class GithubSyncer(Syncer): has_wiki=False, has_discussions=False, ) + self.logger.info("Created new GitHub repository %s", repo.full_name) + try: + _ = repo.get_contents("/") + except GithubException: + self.logger.warning( + "Could not fetch contents of %s, continuing assuming the repo is empty", + repo.name, + ) + + synced_repo = make_synced(repo=repo) + + existing_push_mirrors = self.push_mirrorer.get_matching_mirrors( + repos=[synced_repo] + )[synced_repo.name] + + push_mirror = self.push_mirrorer.mirror_repo( + repo=synced_repo, + existing_push_mirrors=existing_push_mirrors, + config=PushMirrorConfig( + remirror=True, + immediate=True, + ), + ) + if push_mirror is None: + raise SyncError(f"Could not mirror new repository {repo.full_name}") + + needs_mirror = False + + repo.git_tags_url + repo.edit( name=from_repo.name, description=description, @@ -77,7 +136,6 @@ class GithubSyncer(Syncer): if from_repo.default_branch is not None else NotSet, archived=from_repo.archived if from_repo.archived is not None else NotSet, - allow_forking=False, ) self.logger.info("Updated GitHub repository %s", repo.full_name) @@ -86,16 +144,8 @@ class GithubSyncer(Syncer): self.logger.info("Replaced topics on GitHub repository %s", repo.full_name) - if ( - repo.owner.name is None - or from_repo.owner is None - or from_repo.owner.login is None - ): - raise SyncError("received malformed repository") + synced_repo = make_synced(repo=repo) - return SyncedRepository( - new_owner=repo.owner.name, - orig_owner=from_repo.owner.login, - name=repo.name, - clone_url=repo.clone_url, - ) + synced_repo.needs_mirror = needs_mirror + + return synced_repo diff --git a/src/forgesync/mirror.py b/src/forgesync/mirror.py index ddfd5d7..2428ab3 100644 --- a/src/forgesync/mirror.py +++ b/src/forgesync/mirror.py @@ -1,4 +1,5 @@ from collections.abc import Iterable +from dataclasses import dataclass, fields from logging import Logger from typing import Self from pyforgejo import PushMirror, PyforgejoApi @@ -9,20 +10,53 @@ class MirrorError(RuntimeError): pass +@dataclass +class PushMirrorConfig: + interval: str | None = None + remirror: bool | None = None + immediate: bool | None = None + sync_on_commit: bool | None = None + + def overlay(self, other: Self) -> Self: + result = type(self)() + for f in fields(self): + value = ( # pyright: ignore[reportAny] + getattr(other, f.name) + if getattr(other, f.name) is not None + else getattr(self, f.name) + ) + setattr(result, f.name, value) + return result + + def is_valid(self: Self) -> bool: + for f in fields(self): + if getattr(self, f.name) is None: + return False + + return True + + class PushMirrorer: client: PyforgejoApi + config: PushMirrorConfig + mirror_token: str logger: Logger def __init__( self: Self, client: PyforgejoApi, + config: PushMirrorConfig, + mirror_token: str, logger: Logger, ) -> None: self.client = client + self.config = config + self.mirror_token = mirror_token self.logger = logger def get_matching_mirrors( - self: Self, repos: Iterable[SyncedRepository] + self: Self, + repos: Iterable[SyncedRepository], ) -> dict[str, list[PushMirror]]: repo_mirrors: dict[str, list[PushMirror]] = {} @@ -47,27 +81,33 @@ class PushMirrorer: self: Self, repo: SyncedRepository, existing_push_mirrors: list[PushMirror], - interval: str, - remirror: bool, - immediate: bool, - sync_on_commit: bool, - mirror_token: str, + config: PushMirrorConfig, ) -> PushMirror | None: + if not repo.needs_mirror: + return None + + self.logger.info( + f"Setting up mirrors for {repo.orig_owner}/{repo.name} to {repo.new_owner}/{repo.name} at {repo.clone_url}" + ) + + final_config = self.config.overlay(config) + + if not final_config.is_valid(): + raise MirrorError("config is invalid") + def add_push_mirror() -> PushMirror: push_mirror = self.client.repository.repo_add_push_mirror( owner=repo.orig_owner, repo=repo.name, - interval=interval, + interval=final_config.interval, remote_address=repo.clone_url, remote_username=repo.new_owner, - remote_password=mirror_token, - sync_on_commit=sync_on_commit, + remote_password=self.mirror_token, + sync_on_commit=final_config.sync_on_commit, use_ssh=False, ) - self.logger.info( - f"Created push mirror for {repo.orig_owner}/{repo.name} to {repo.new_owner}/{repo.name} at {repo.clone_url}" - ) + self.logger.info("Created push mirror") return push_mirror @@ -77,16 +117,14 @@ class PushMirrorer: if push_mirror.remote_name is None: raise MirrorError("missing remote name") - if remirror: + if final_config.remirror: self.client.repository.repo_delete_push_mirror( owner=repo.orig_owner, repo=repo.name, name=push_mirror.remote_name, ) - self.logger.info( - f"Removed old push mirror for {repo.orig_owner}/{repo.name} to {repo.new_owner}/{repo.name} at {repo.clone_url}" - ) + self.logger.info("Removed old push mirror") new_push_mirror = add_push_mirror() @@ -94,7 +132,7 @@ class PushMirrorer: new_push_mirror = add_push_mirror() if new_push_mirror is not None: - if immediate: + if final_config.immediate: self.client.repository.repo_push_mirror_sync( owner=repo.orig_owner, repo=repo.name, @@ -105,11 +143,7 @@ class PushMirrorer: def mirror_repos( self: Self, synced_repos: Iterable[SyncedRepository], - interval: str, - remirror: bool, - immediate: bool, - sync_on_commit: bool, - mirror_token: str, + config: PushMirrorConfig, ) -> list[PushMirror]: new_push_mirrors: list[PushMirror] = [] @@ -119,11 +153,7 @@ class PushMirrorer: new_push_mirror = self.mirror_repo( repo=synced_repo, existing_push_mirrors=matching_mirrors[synced_repo.name], - interval=interval, - remirror=remirror, - immediate=immediate, - sync_on_commit=sync_on_commit, - mirror_token=mirror_token, + config=config, ) if new_push_mirror is not None: diff --git a/src/forgesync/sync.py b/src/forgesync/sync.py index af33100..1fa46ce 100644 --- a/src/forgesync/sync.py +++ b/src/forgesync/sync.py @@ -1,15 +1,23 @@ from abc import ABC, abstractmethod from dataclasses import dataclass +from enum import StrEnum from typing import Self from pyforgejo import Repository as ForgejoRepository +class Destination(StrEnum): + GITHUB = "github" + FORGEJO = "forgejo" + + @dataclass class SyncedRepository: new_owner: str orig_owner: str name: str clone_url: str + destination: Destination + needs_mirror: bool class Syncer(ABC):