100 lines
2.4 KiB
Python
100 lines
2.4 KiB
Python
|
"""
|
||
|
|
Repository migration script
|
||
|
|
|
||
|
|
Facilitates the migration from Github to Forgejo by importing
|
||
|
|
all public/private repositories owned by the user attached to
|
||
|
|
the personal token included.
|
||
|
|
|
||
|
|
Expects a single argument that points to a json configuration
|
||
|
|
shaped like Config.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# /// script
|
||
|
|
# dependencies = [
|
||
|
|
# "httpx",
|
||
|
|
# "pydantic"
|
||
|
|
# ]
|
||
|
|
# ///
|
||
|
|
|
||
|
|
import sys
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pydantic
|
||
|
|
|
||
|
|
|
||
|
|
class Config(pydantic.BaseModel):
|
||
|
|
gh_user: str
|
||
|
|
gh_token: str
|
||
|
|
gh_host: str
|
||
|
|
gh_api_host: str
|
||
|
|
|
||
|
|
forgejo_host: str
|
||
|
|
forgejo_user: str
|
||
|
|
forgejo_token: str
|
||
|
|
|
||
|
|
|
||
|
|
def get_config(path: str) -> Config:
|
||
|
|
return Config.parse_file(path)
|
||
|
|
|
||
|
|
|
||
|
|
def get_gh_client(config: Config):
|
||
|
|
return httpx.Client(
|
||
|
|
headers={
|
||
|
|
"Authorization": f"Bearer {config.gh_token}",
|
||
|
|
"X-Github-Api-Version": "2022-11-28",
|
||
|
|
},
|
||
|
|
base_url=config.gh_api_host,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def get_forgejo_client(config: Config):
|
||
|
|
return httpx.Client(
|
||
|
|
headers={"Authorization": f"token {config.forgejo_token}"},
|
||
|
|
base_url=config.forgejo_host,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def get_repositories_from_github(config: Config) -> list[str]:
|
||
|
|
"""Gets all repositories for given user."""
|
||
|
|
|
||
|
|
with get_gh_client(config) as client:
|
||
|
|
response = client.get(
|
||
|
|
f"/user/repos", params={"per_page": 100, "affiliation": "owner"}
|
||
|
|
)
|
||
|
|
|
||
|
|
for repository in response.json():
|
||
|
|
full_name = repository["full_name"]
|
||
|
|
|
||
|
|
if not full_name.startswith(config.gh_user):
|
||
|
|
print(f"Skipped {full_name}: not owned.")
|
||
|
|
continue
|
||
|
|
|
||
|
|
yield full_name
|
||
|
|
|
||
|
|
|
||
|
|
def migrate_to_forgejo(repository_name: str, config: Config):
|
||
|
|
with get_forgejo_client(config) as client:
|
||
|
|
response = client.post(
|
||
|
|
"/repos/migrate",
|
||
|
|
data={
|
||
|
|
"auth_username": config.gh_user,
|
||
|
|
"auth_token": config.gh_token,
|
||
|
|
"clone_addr": f"{config.gh_host}/{repository_name}.git",
|
||
|
|
"repo_name": repository_name.split("/")[1],
|
||
|
|
},
|
||
|
|
timeout=None,
|
||
|
|
)
|
||
|
|
|
||
|
|
if response.status_code == 409:
|
||
|
|
print(f"Did not migrate {repository_name}: already exists")
|
||
|
|
elif response.status_code == 201:
|
||
|
|
print(f"Migrated {repository_name}")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
config = get_config(sys.argv[1])
|
||
|
|
|
||
|
|
for repo in get_repositories_from_github(config):
|
||
|
|
print(f"Migrating {repo}...")
|
||
|
|
migrate_to_forgejo(repo, config)
|