feat: add Forgejo saas wrapper
Mirrors the Bitbucket wrapper against the Forgejo REST API v1: token auth headers, repository existence check, repository creation under the authenticated user or an organization. No instance URL is hardcoded — Forgejo is self-hosted, so every operation takes a host parameter. Exposes both ssh_clone_url and https_clone_url (HTTPS needed in CI without SSH host keys).
This commit is contained in:
parent
73c32fdee0
commit
e47de33caf
1 changed files with 98 additions and 0 deletions
98
src/byteb4rb1e/utils/saas/forgejo.py
Normal file
98
src/byteb4rb1e/utils/saas/forgejo.py
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Forgejo REST API v1 wrapper.
|
||||
|
||||
Thin layer over http.py for Forgejo-specific operations:
|
||||
|
||||
- Token authentication
|
||||
- Repository existence checks
|
||||
- Repository creation under the authenticated user or an organization
|
||||
- SSH and HTTPS clone URL construction
|
||||
|
||||
Unlike Bitbucket (one global SaaS instance), Forgejo is self-hosted,
|
||||
so every operation takes a *host* parameter instead of baking any
|
||||
specific instance in.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from byteb4rb1e.utils.http import client as http_client
|
||||
|
||||
|
||||
def api_url(host: str) -> str:
|
||||
"""Return the API base URL for a Forgejo instance."""
|
||||
return f"https://{host}/api/v1"
|
||||
|
||||
|
||||
def http_headers(token: str) -> Dict[str, str]:
|
||||
"""Construct Forgejo API headers with token auth."""
|
||||
return {
|
||||
"Authorization": f"token {token}",
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
|
||||
def repository_exists(
|
||||
host: str,
|
||||
owner: str,
|
||||
repo_slug: str,
|
||||
token: str,
|
||||
) -> bool:
|
||||
"""Check whether a repository exists under the owner."""
|
||||
url = f"{api_url(host)}/repos/{owner}/{repo_slug}"
|
||||
resp = http_client.get(url, headers=http_headers(token))
|
||||
return bool(resp.status_code == 200)
|
||||
|
||||
|
||||
def create_repository(
|
||||
host: str,
|
||||
repo_slug: str,
|
||||
token: str,
|
||||
org: Optional[str] = None,
|
||||
description: str = "",
|
||||
is_private: bool = True,
|
||||
) -> http_client.HttpResponse:
|
||||
"""Create a new repository on the Forgejo instance.
|
||||
|
||||
When *org* is given the repository is created in that
|
||||
organization, otherwise under the authenticated user.
|
||||
|
||||
Returns the API response. Caller should check status_code == 201
|
||||
for success.
|
||||
"""
|
||||
if org:
|
||||
url = f"{api_url(host)}/orgs/{org}/repos"
|
||||
else:
|
||||
url = f"{api_url(host)}/user/repos"
|
||||
body: Dict[str, Any] = {
|
||||
"name": repo_slug,
|
||||
"private": is_private,
|
||||
"description": description,
|
||||
}
|
||||
return http_client.post(
|
||||
url,
|
||||
data=json.dumps(body).encode("utf-8"),
|
||||
headers=http_headers(token),
|
||||
)
|
||||
|
||||
|
||||
def ssh_clone_url(
|
||||
host: str,
|
||||
owner: str,
|
||||
repo_slug: str,
|
||||
) -> str:
|
||||
"""Return the SSH clone URL for a Forgejo repository."""
|
||||
return f"git@{host}:{owner}/{repo_slug}.git"
|
||||
|
||||
|
||||
def https_clone_url(
|
||||
host: str,
|
||||
owner: str,
|
||||
repo_slug: str,
|
||||
) -> str:
|
||||
"""Return the HTTPS clone URL for a Forgejo repository.
|
||||
|
||||
Preferred in CI environments without SSH host keys.
|
||||
"""
|
||||
return f"https://{host}/{owner}/{repo_slug}.git"
|
||||
Loading…
Add table
Add a link
Reference in a new issue