96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
import requests
|
|
import base64
|
|
from urllib.parse import urljoin
|
|
|
|
class GiteaConnector:
|
|
def __init__(self, repo_url, user, key):
|
|
self.repo_url = repo_url
|
|
self.user = user
|
|
self.key = key
|
|
|
|
self.header = {
|
|
"accept": "application/json",
|
|
"Authorization": f"token {self.key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
## Create an issue in the repo
|
|
def create_issue(self,title,content):
|
|
url_suffix = "issues"
|
|
url = urljoin(self.repo_url, url_suffix)
|
|
|
|
issue_data = {
|
|
"title": title,
|
|
"body": content,
|
|
"assignees": ["fred.boniface"],
|
|
"labels": ["bug", "urgent"],
|
|
}
|
|
|
|
response = requests.post(url, headers=self.header, json=issue_data)
|
|
if response.status_code == 201:
|
|
print("Succesfully creates issue: ", response.json())
|
|
return True
|
|
else:
|
|
print("Failed to create issue: ", response.status_code, response.text)
|
|
return False
|
|
|
|
## Download the 'gw' pis file from the repo
|
|
def download_pis_file(self):
|
|
url = urljoin(self.repo_url, 'contents/pis/gw.yaml')
|
|
response = requests.get(url, headers=self.header)
|
|
if response.status_code == 200:
|
|
file_content = base64.b64decode(response.json()['content']).decode('utf-8')
|
|
print("File downloaded successfully")
|
|
return file_content
|
|
else:
|
|
print("Failed to download file: ", response.status_code, response.text)
|
|
return ""
|
|
|
|
## Create new branch in the repo
|
|
def create_branch(self, branch_name):
|
|
BASE_BRANCH = "main"
|
|
base_branch_url = urljoin(self.repo_url, f"branch/{BASE_BRANCH}")
|
|
branch_response = requests.get(base_branch_url, headers=self.header)
|
|
|
|
if branch_response.status_code == 200:
|
|
base_sha = branch_response.json()['commit']['id']
|
|
|
|
create_branch_url = urljoin(self.repo_url, "git/refs")
|
|
branch_data = {
|
|
"ref": f"refs/heads/{branch_name}",
|
|
"sha": base_sha,
|
|
}
|
|
|
|
create_branch_response = requests.post(create_branch_url, headers=self.header, data=json.dumps(branch_data))
|
|
if create_branch_response.status_code == 201:
|
|
print(f"Branch {branch_name} created successfully")
|
|
return True
|
|
else:
|
|
print(f"Failed to create branch: ", branch_name, create_branch_response.status_code, create_branch_response.text)
|
|
return False
|
|
|
|
else:
|
|
print("Failed to retreive branch SHA: ", branch_response.status_code, branch_response.text)
|
|
return False
|
|
|
|
## Create a new file in the repo
|
|
def create_pis_file(self, branch, filename, file_content):
|
|
file_path = f"pis/{filename}"
|
|
encoded_content = base64.encode(file_content.encode('utf-8'))
|
|
|
|
create_file_url = url_multijoin(self.repo_url, f'contents/{file_path}')
|
|
|
|
file_data = {
|
|
"content": encoded_content,
|
|
"message": "Create new PIS File",
|
|
"branch": branch,
|
|
}
|
|
|
|
create_file_response = requests.post(create_file_url, headers=self.header, data=json.dumps(file_data))
|
|
|
|
if create_file_response.status_code == 201:
|
|
print("File created successfully")
|
|
return True
|
|
else:
|
|
print("File creation failed.", create_file_response.status_code, create_file_response.text)
|
|
return False |