gitea-github-sync/bridge/webgit.py

186 lines
5.1 KiB
Python
Raw Normal View History

2024-10-09 03:17:22 -05:00
from typing import Any
from dataclasses import dataclass, field
import requests
from requests import Request, Session
from requests.exceptions import HTTPError
from flask import abort
@dataclass
class WebgitClient:
"""
A quite thin wrapper around various code forges' REST APIs.
Designed to be subclassed.
"""
api_token: str
headers: Any = field(init=False)
2024-10-09 05:16:33 -05:00
api_prefix: str = field(init=False)
2024-10-09 03:17:22 -05:00
def _post_request(self, request_obj):
try:
request_obj.raise_for_status()
except HTTPError as e:
print("An exception occured: {}({})".format(
type(e).__name__, e
))
abort(500)
def _request_wrapper(self, method, *args, **kwargs):
s = Session()
r = Request(method, *args, **kwargs, headers=self.headers)
prepped = s.prepare_request(r)
settings = s.merge_environment_settings(prepped.url, {}, None, None, None)
r = s.send(prepped, **settings)
self._post_request(r)
return r
def post(self, *args, **kwargs):
return self._request_wrapper("POST", *args, **kwargs)
def patch(self, *args, **kwargs):
return self._request_wrapper("PATCH", *args, **kwargs)
2024-10-09 05:16:33 -05:00
def create_repo(self, name, description):
return self.post(
self.api_prefix + "/user/repos",
json={
"name": name,
"description": description,
},
)
def create_issue(self, owner, repo_name, title, body):
return self.post(
self.api_prefix + "/repos/{}/{}/issues".format(
owner,
repo_name,
),
json={
"title": title,
"body": body,
},
)
def leave_comment_on_issue_by_number(self, owner, repo_name, issue_number, body):
return self.post(
self.api_prefix + "/repos/{}/{}/issues/{}/comments".format(
owner,
repo_name,
issue_number,
),
json={
"body": body,
},
)
def leave_comment_on_issue_by_url(self, url, body):
return self.post(
url + "/comments",
json={
"body": body,
},
)
def close_issue_by_number(self, owner, repo_name, issue_number):
return self.patch(
self.api_prefix + "/repos/{}/{}/issues/{}".format(
owner,
repo_name,
issue_number,
),
json={
"state": "closed",
},
)
def close_issue_by_url(self, url):
return self.patch(
url,
json={
"state": "closed",
},
)
2024-10-09 03:17:22 -05:00
@dataclass
class Github(WebgitClient):
"""
A quite thin wrapper around Github's REST API.
"""
def __post_init__(self):
self.headers = {
"Accept": "application/vnd.github+json",
"Authorization": "token " + self.api_token,
"X-GitHub-Api-Version": "2022-11-28",
}
2024-10-09 05:16:33 -05:00
self.api_prefix = "https://api.github.com"
def create_webhook(self, owner, repo_name, http_endpoint, events: list[str]):
return self.post(
self.api_prefix + "/repos/{}/{}/hooks".format(
owner,
repo_name,
),
json={
"name": "web",
"config": {
"url": http_endpoint,
"content_type": "json",
},
"events": events,
},
)
2024-10-09 03:17:22 -05:00
@dataclass
class Gitea(WebgitClient):
"""
A quite thin wrapper around Gitea's REST API.
"""
2024-10-09 05:16:33 -05:00
instance_name: str
2024-10-09 03:17:22 -05:00
def __post_init__(self):
self.headers = {
"Authorization": "token " + self.api_token,
}
2024-10-09 05:16:33 -05:00
self.api_prefix = "https://{}/api/v1".format(self.instance_name)
def add_push_target(self, owner, repo_name, address, username, password):
return self.post(
self.api_prefix + "/repos/{}/{}/push_mirrors".format(
owner, repo_name
),
json={
"interval": "8h0m0s",
"remote_address": address,
"remote_password": password,
"remote_username": username,
"sync_on_commit": True,
},
)
def force_push_target(self, owner, repo_name):
return self.post(
self.api_prefix + "/repos/{}/{}/push_mirrors-sync".format(
owner,
repo_name
),
)
def create_webhook(self, owner, repo_name, http_endpoint, events: list[str]):
return self.post(
self.api_prefix + "/repos/{}/{}/hooks".format(
owner,
repo_name,
),
json={
"config": {
"url": http_endpoint,
"content_type": "json",
},
"events": events,
"type": "gitea",
},
)