import os import sys import requests import pdb class Gitea: name = 'Gitea API wrapper' def __init__(self, api_base_url=None, access_token=None, session=None): self.__gitea_config_path = "config/gitea.txt" is_setup = self.__check_setup(self) if is_setup: self.__api_base_url = self.__get_parameter("api_base_url", self.__gitea_config_path) self.__access_token = self.__get_parameter("access_token", self.__gitea_config_path) else: self.__api_base_url, self.__access_token = self.__setup(self) if session: self.session = session else: self.session = requests.Session() def register_user(self, email, username, passwd): data = {'email':email, 'full_name':username, 'login_name':username, 'must_change_password':True, 'password':passwd, 'restricted':False, 'send_notify':True, 'source_id':'0', 'username':username, 'visibility':'public' } endpoint = self._Gitea__api_base_url + f'/api/v1/admin/users?access_token={self._Gitea__access_token}' response = self.__api_request('POST', endpoint, data) is_registered = response.ok message = response.json()['message'] return (is_registered, message) def list_users(self): data = { } endpoint = self._Gitea__api_base_url + f'/api/v1/admin/users?token={self._Gitea__access_token}' response = self.__api_request('GET', endpoint, data) response_ok = response.ok user_list = response.json() return (response_ok, user_list) @staticmethod def __check_setup(self): is_setup = False if not os.path.isfile(self.__gitea_config_path): print(f"File {self.__gitea_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') api_base_url = input("Gitea API base url, in ex. 'https://yourgitea.instance': ") access_token = input("Gitea access token: ") if not os.path.exists(self.__gitea_config_path): with open(self.__gitea_config_path, 'w'): pass print(f"{self.__gitea_config_path} created!") with open(self.__gitea_config_path, 'a') as the_file: print("Writing gitea parameters to " + self.__gitea_config_path) the_file.write(f'api_base_url: {api_base_url}\n'+f'access_token: {access_token}\n') return (api_base_url, 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): try: response = self.session.request(method, url = endpoint, data = data) except Exception as e: raise GiteaNetworkError(f"Could not complete request: {e}") if response is None: raise GiteaIllegalArgumentError("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 = GiteaNotFoundError if not error_msg: error_msg = 'Endpoint not found.' # this is for compatibility with older versions # which raised GiteaAPIError('Endpoint not found.') # on any 404 elif response.status_code == 401: ex_type = GiteaUnauthorizedError elif response.status_code == 422: return response elif response.status_code == 500: ex_type = GiteaInternalServerError elif response.status_code == 502: ex_type = GiteaBadGatewayError elif response.status_code == 503: ex_type = GiteaServiceUnavailableError elif response.status_code == 504: ex_type = GiteaGatewayTimeoutError elif response.status_code >= 500 and \ response.status_code <= 511: ex_type = GiteaServerError else: ex_type = GiteaAPIError raise ex_type( 'Gitea API returned error', response.status_code, response.reason, error_msg) else: return response ## # Exceptions ## class GiteaError(Exception): """Base class for Gitea.py exceptions""" class GiteaIOError(IOError, GiteaError): """Base class for Gitea.py I/O errors""" class GiteaNetworkError(GiteaIOError): """Raised when network communication with the server fails""" pass class GiteaAPIError(GiteaError): """Raised when the gitea API generates a response that cannot be handled""" pass class GiteaServerError(GiteaAPIError): """Raised if the Server is malconfigured and returns a 5xx error code""" pass class GiteaInternalServerError(GiteaServerError): """Raised if the Server returns a 500 error""" pass class GiteaBadGatewayError(GiteaServerError): """Raised if the Server returns a 502 error""" pass class GiteaServiceUnavailableError(GiteaServerError): """Raised if the Server returns a 503 error""" pass class GiteaGatewayTimeoutError(GiteaServerError): """Raised if the Server returns a 504 error""" pass class GiteaNotFoundError(GiteaAPIError): """Raised when the gitea API returns a 404 Not Found error""" pass class GiteaUnauthorizedError(GiteaAPIError): """Raised when the gitea 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