527 líneas
17 KiB
Python
527 líneas
17 KiB
Python
import os
|
|
import sys
|
|
import requests
|
|
from getpass import getpass
|
|
from requests.models import urlencode
|
|
import json
|
|
import pdb
|
|
|
|
###
|
|
# Dict helper class.
|
|
# Defined at top level so it can be pickled.
|
|
###
|
|
class AttribAccessDict(dict):
|
|
def __getattr__(self, attr):
|
|
if attr in self:
|
|
return self[attr]
|
|
else:
|
|
raise AttributeError("Attribute not found: " + str(attr))
|
|
|
|
def __setattr__(self, attr, val):
|
|
if attr in self:
|
|
raise AttributeError("Attribute-style access is read only")
|
|
super(AttribAccessDict, self).__setattr__(attr, val)
|
|
|
|
class Forgejo:
|
|
|
|
name = 'Forgejo API wrapper'
|
|
|
|
def __init__(self, api_base_url=None, access_token=None, session=None):
|
|
|
|
self.__forgejo_config_path = "config/forgejo.txt"
|
|
|
|
is_setup = self.__check_setup(self)
|
|
|
|
if is_setup:
|
|
|
|
self.api_base_url = self.__get_parameter("api_base_url", self.__forgejo_config_path)
|
|
self.access_token = self.__get_parameter("access_token", self.__forgejo_config_path)
|
|
|
|
else:
|
|
|
|
self.api_base_url, self.access_token = self.__setup(self)
|
|
|
|
if session:
|
|
self.session = session
|
|
else:
|
|
self.session = requests.Session()
|
|
|
|
def admin_users_create(self, email, username, passwd, full_name=None, must_change_password=True, restricted=False, send_notify=True, source_id=0, visibility="public"):
|
|
|
|
data = {
|
|
'email':email,
|
|
'full_name':username,
|
|
'login_name':username,
|
|
'must_change_password':must_change_password,
|
|
'password':passwd,
|
|
'restricted':restricted,
|
|
'send_notify':send_notify,
|
|
'source_id':source_id,
|
|
'username':username,
|
|
'visibility':visibility
|
|
}
|
|
|
|
endpoint = self.api_base_url + '/api/v1/admin/users?token={0}'.format(self.access_token)
|
|
|
|
response = self.__api_request('POST', endpoint, data)
|
|
|
|
registered = response.ok
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return (registered, response)
|
|
|
|
def admin_users_list(self, page=None, limit=None):
|
|
|
|
params = dict()
|
|
if page != None:
|
|
params['page'] = page
|
|
if limit != None:
|
|
params['limit'] = limit
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/admin/users?{0}'.format(formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def notifications_new(self):
|
|
|
|
endpoint = self.api_base_url + '/api/v1/notifications/new?token={0}'.format(self.access_token)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
###
|
|
### repository
|
|
###
|
|
|
|
def repos_get_repo(self, owner, repo):
|
|
|
|
params = dict()
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}'.format(owner, repo, formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repos_issues_search(self, owner, state=None, labels=None, q=None, milestones=None, priority_repo_id=None, issue_type=None, since=None, before=None, assigned=None,
|
|
created=None, mentioned=None, review_requested=None, team=None, page=None, limit=None):
|
|
|
|
params = dict()
|
|
|
|
if state == None:
|
|
params['state'] = 'open'
|
|
else:
|
|
params['state'] = state
|
|
if labels != None:
|
|
params['labels'] = labels
|
|
if milestones != None:
|
|
params['milestones'] = milestones
|
|
if q != None:
|
|
params['q'] = q
|
|
if priority_repo_id != None:
|
|
params['priority_repo_id'] = priority_repo_id
|
|
if issue_type != None:
|
|
params['type'] = issue_type
|
|
if since != None:
|
|
params['since'] = since
|
|
if before != None:
|
|
params['before'] = before
|
|
if assigned != None:
|
|
params['assigned'] = assigned
|
|
if created != None:
|
|
params['created'] = created
|
|
if mentioned != None:
|
|
params['mentioned'] = mentioned
|
|
if review_requested != None:
|
|
params['review_requested'] = review_requested
|
|
params['owner'] = owner
|
|
if team != None:
|
|
params['team'] = team
|
|
if page != None:
|
|
params['page'] = page
|
|
if limit != None:
|
|
params['limit'] = limit
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/issues/search?{0}'.format(formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repos_owner_repo_issues(self, owner, repo, state=None, labels=None, q=None, issue_type=None, milestones=None, since=None, before=None, created_by=None, assigned_by=None, mentioned_by=None, page=None, limit=None):
|
|
"""
|
|
List a repository's issues
|
|
If since or before are specified, they must have following format: in. ex. 2022-08-13T08:09:07+02:00
|
|
"""
|
|
params = dict()
|
|
|
|
if state == None:
|
|
params['state'] = 'open'
|
|
else:
|
|
params['state'] = state
|
|
params['labels'] = labels
|
|
if q != None:
|
|
params['q'] = q
|
|
params['issue_type'] = issue_type
|
|
params['milestones'] = milestones
|
|
if since != None:
|
|
params['since'] = since
|
|
if before != None:
|
|
params['before'] = before
|
|
if created_by != None:
|
|
params['created_by'] = created_by
|
|
if assigned_by != None:
|
|
params['assigned_by'] = assigned_by
|
|
if mentioned_by != None:
|
|
params['mentioned_by'] = mentioned_by
|
|
params['page'] = page
|
|
params['limit'] = limit
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/issues?{2}'.format(owner, repo, formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repos_owner_repo_issues_pinned(self, owner, repo):
|
|
"""
|
|
List a repo's pinned isssues
|
|
"""
|
|
params = dict()
|
|
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/issues?{2}'.format(owner, repo, formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repos_owner_repo_issues_comments(self, owner, repo, since=None, before=None, page=None, limit=None):
|
|
"""
|
|
if since or before are specified, they must have following format: in. ex. 2022-08-13T08:09:07+02:00
|
|
"""
|
|
params = dict()
|
|
if since != None:
|
|
params['since'] = since
|
|
if before != None:
|
|
params['before'] = before
|
|
params['page'] = page
|
|
params['limit'] = limit
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/issues/comments?{2}'.format(owner, repo, formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)#, data)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repo_owner_get_metadata(self, owner, repo, filepath):
|
|
"""
|
|
Gets the metadata and contents (if a file) of an entry in a repository, or list of entries if a dir
|
|
"""
|
|
|
|
params = dict()
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/contents/{2}?{3}'.format(owner, repo, filepath, formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repo_owner_create_file(self, owner, repo, filepath, author_email, author_name, branch, content, message):
|
|
"""
|
|
Create a file in a repository
|
|
"""
|
|
|
|
content = json.dumps(content.decode('UTF-8')).replace('"','')
|
|
|
|
data = {"author":{
|
|
"email":author_email,
|
|
"name":author_name
|
|
},
|
|
"branch":branch,
|
|
"content":content,
|
|
"message":message
|
|
}
|
|
|
|
params = dict()
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/contents/{2}?{3}'.format(owner, repo, filepath, formatted_params)
|
|
|
|
response = self.__api_request('POST', endpoint, data)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repo_owner_update_file(self, owner, repo, filepath, author_email, author_name, branch, content, message, sha):
|
|
"""
|
|
Update a file in a repository
|
|
"""
|
|
content = json.dumps(content.decode('UTF-8')).replace('"','')
|
|
|
|
data = {"author":{
|
|
"email":author_email,
|
|
"name":author_name
|
|
},
|
|
"branch":branch,
|
|
"content":content,
|
|
"message":message,
|
|
"sha":sha
|
|
}
|
|
|
|
params = dict()
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/contents/{2}?{3}'.format(owner, repo, filepath, formatted_params)
|
|
|
|
response = self.__api_request('PUT', endpoint, data)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def repo_owner_delete_file(self, owner, repo, filepath, author_email, author_name, branch, message, sha):
|
|
"""
|
|
Delete a file in a repository
|
|
"""
|
|
data = {"author":{
|
|
"email":author_email,
|
|
"name":author_name
|
|
},
|
|
"branch":branch,
|
|
"commiter":{
|
|
"email":author_email,
|
|
"name":author_name
|
|
},
|
|
"message":message,
|
|
"new_branch":branch,
|
|
'sha':sha
|
|
}
|
|
|
|
params = dict()
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/contents/{2}?{3}'.format(owner, repo, filepath, formatted_params)
|
|
|
|
response = self.__api_request('DELETE', endpoint, data)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
def user(self):
|
|
"""
|
|
Get the authenticated user
|
|
"""
|
|
|
|
params = dict()
|
|
params['token'] = self.access_token
|
|
formatted_params = urlencode(params)
|
|
|
|
endpoint = self.api_base_url + '/api/v1/user?{0}'.format(formatted_params)
|
|
|
|
response = self.__api_request('GET', endpoint)
|
|
|
|
response = self.__json_allow_dict_attrs(response.json())
|
|
|
|
return response
|
|
|
|
@staticmethod
|
|
def __check_setup(self):
|
|
|
|
is_setup = False
|
|
|
|
if not os.path.isfile(self.__forgejo_config_path):
|
|
print(f"File {self.__forgejo_config_path} not found, running setup.")
|
|
else:
|
|
is_setup = True
|
|
|
|
return is_setup
|
|
|
|
@staticmethod
|
|
def __setup(self):
|
|
|
|
if not os.path.exists('config'):
|
|
os.makedirs('config')
|
|
|
|
self.api_base_url = input("Forgejo API base url, in ex. 'https://yourforgejo.instance': ")
|
|
self.access_token = getpass("Forgejo access token: ")
|
|
|
|
if not os.path.exists(self.__forgejo_config_path):
|
|
with open(self.__forgejo_config_path, 'w'): pass
|
|
print(f"{self.__forgejo_config_path} created!")
|
|
|
|
with open(self.__forgejo_config_path, 'a') as the_file:
|
|
print("Writing forgejo parameters to " + self.__forgejo_config_path)
|
|
the_file.write(f'api_base_url: {self.api_base_url}\n'+f'access_token: {self.access_token}')
|
|
|
|
return (self.api_base_url, self.access_token)
|
|
|
|
@staticmethod
|
|
def __get_parameter(parameter, file_path ):
|
|
|
|
with open( file_path ) as f:
|
|
for line in f:
|
|
if line.startswith( parameter ):
|
|
return line.replace(parameter + ":", "").strip()
|
|
|
|
print(f'{file_path} Missing parameter {parameter}')
|
|
sys.exit(0)
|
|
|
|
def __api_request(self, method, endpoint, data={}):
|
|
|
|
headers = {
|
|
'Accept':'application/json',
|
|
'Content-Type':'application/json'
|
|
}
|
|
|
|
response = None
|
|
|
|
try:
|
|
|
|
response = self.session.request(method, url = endpoint, headers=headers, json=data)
|
|
|
|
except Exception as e:
|
|
|
|
raise ForgejoNetworkError(f"Could not complete request: {e}")
|
|
|
|
if response is None:
|
|
|
|
raise ForgejoIllegalArgumentError("Illegal request.")
|
|
|
|
if not response.ok:
|
|
|
|
try:
|
|
if isinstance(response, dict) and 'error' in response:
|
|
error_msg = response['error']
|
|
elif isinstance(response, str):
|
|
error_msg = response
|
|
else:
|
|
error_msg = None
|
|
except ValueError:
|
|
error_msg = None
|
|
|
|
if response.status_code == 404:
|
|
ex_type = ForgejoNotFoundError
|
|
if not error_msg:
|
|
error_msg = 'Endpoint not found.'
|
|
# this is for compatibility with older versions
|
|
# which raised ForgejoAPIError('Endpoint not found.')
|
|
# on any 404
|
|
elif response.status_code == 401:
|
|
ex_type = ForgejoUnauthorizedError
|
|
elif response.status_code == 422:
|
|
return response
|
|
elif response.status_code == 500:
|
|
ex_type = ForgejoInternalServerError
|
|
elif response.status_code == 502:
|
|
ex_type = ForgejoBadGatewayError
|
|
elif response.status_code == 503:
|
|
ex_type = ForgejoServiceUnavailableError
|
|
elif response.status_code == 504:
|
|
ex_type = ForgejoGatewayTimeoutError
|
|
elif response.status_code >= 500 and \
|
|
response.status_code <= 511:
|
|
ex_type = ForgejoServerError
|
|
else:
|
|
ex_type = ForgejoAPIError
|
|
|
|
raise ex_type(
|
|
'Forgejo API returned error',
|
|
response.status_code,
|
|
response.reason,
|
|
error_msg)
|
|
|
|
else:
|
|
|
|
return response
|
|
|
|
@staticmethod
|
|
def __json_allow_dict_attrs(json_object):
|
|
"""
|
|
Makes it possible to use attribute notation to access a dicts
|
|
elements, while still allowing the dict to act as a dict.
|
|
"""
|
|
if isinstance(json_object, dict):
|
|
return AttribAccessDict(json_object)
|
|
return json_object
|
|
|
|
##
|
|
# Exceptions
|
|
##
|
|
class ForgejoError(Exception):
|
|
"""Base class for Forgejo.py exceptions"""
|
|
|
|
class ForgejoIOError(IOError, ForgejoError):
|
|
"""Base class for Forgejo.py I/O errors"""
|
|
|
|
class ForgejoNetworkError(ForgejoIOError):
|
|
"""Raised when network communication with the server fails"""
|
|
pass
|
|
class ForgejoAPIError(ForgejoError):
|
|
"""Raised when the forgejo API generates a response that cannot be handled"""
|
|
pass
|
|
class ForgejoServerError(ForgejoAPIError):
|
|
"""Raised if the Server is malconfigured and returns a 5xx error code"""
|
|
pass
|
|
class ForgejoInternalServerError(ForgejoServerError):
|
|
"""Raised if the Server returns a 500 error"""
|
|
pass
|
|
|
|
class ForgejoBadGatewayError(ForgejoServerError):
|
|
"""Raised if the Server returns a 502 error"""
|
|
pass
|
|
|
|
class ForgejoServiceUnavailableError(ForgejoServerError):
|
|
"""Raised if the Server returns a 503 error"""
|
|
pass
|
|
class ForgejoGatewayTimeoutError(ForgejoServerError):
|
|
"""Raised if the Server returns a 504 error"""
|
|
pass
|
|
class ForgejoNotFoundError(ForgejoAPIError):
|
|
"""Raised when the forgejo API returns a 404 Not Found error"""
|
|
pass
|
|
|
|
class ForgejoUnauthorizedError(ForgejoAPIError):
|
|
"""Raised when the forgejo API returns a 401 Unauthorized error
|
|
|
|
This happens when an OAuth token is invalid or has been revoked,
|
|
or when trying to access an endpoint that can't be used without
|
|
authentication without providing credentials."""
|
|
pass
|