helvetica/forgesync
1
0
Fork 0
Code Activity

fix github syncing

Browse source
This commit is contained in:
Lukas Wurzinger 2025-05-02 00:08:15 +02:00
parent e42a8a9185
commit a3c7618824
No known key found for this signature in database
5 changed files with 169 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

46
src/forgesync/cli.py
View file

@ -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)

19
src/forgesync/forgejo.py
View file

@ -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,
)

82
src/forgesync/github.py
View file

@ -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

84
src/forgesync/mirror.py
View file

@ -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:

8
src/forgesync/sync.py
View file

@ -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):