diff --git a/Gitea.py b/Gitea.py index 213a5a3..d767a5d 100644 --- a/Gitea.py +++ b/Gitea.py @@ -1,6 +1,7 @@ import os import sys import requests +from requests.models import urlencode class Gitea: @@ -14,33 +15,33 @@ class Gitea: 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) + 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) + 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): + def admin_users_create(self, email, username, passwd, full_name=None, must_change_password=True, restricted=False, send_notify=True, source_id='0', visibility='private'): data = {'email':email, 'full_name':username, 'login_name':username, - 'must_change_password':True, + 'must_change_password':must_change_password, 'password':passwd, - 'restricted':False, - 'send_notify':True, - 'source_id':'0', + 'restricted':restricted, + 'send_notify':send_notify, + 'source_id':source_id, 'username':username, - 'visibility':'public' + 'visibility':visibility } - endpoint = self._Gitea__api_base_url + f'/api/v1/admin/users?access_token={self._Gitea__access_token}' + endpoint = self.api_base_url + '/api/v1/admin/users?access_token={0}'.format(self.access_token) response = self.__api_request('POST', endpoint, data) @@ -50,14 +51,19 @@ class Gitea: return (is_registered, message) - def list_users(self): + def admin_users_list(self, page=None, limit=None): - data = { - } + 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._Gitea__api_base_url + f'/api/v1/admin/users?token={self._Gitea__access_token}' + endpoint = self.api_base_url + '/api/v1/admin/users?{0}'.format(formatted_params) - response = self.__api_request('GET', endpoint, data) + response = self.__api_request('GET', endpoint) response_ok = response.ok @@ -67,12 +73,9 @@ class Gitea: def notifications_new(self): - data = { - } + endpoint = self.api_base_url + '/api/v1/notifications/new?token={0}'.format(self.access_token) - endpoint = self._Gitea__api_base_url + f'/api/v1/notifications/new?token={self._Gitea__access_token}' - - response = self.__api_request('GET', endpoint, data) + response = self.__api_request('GET', endpoint) response_ok = response.ok @@ -80,14 +83,90 @@ class Gitea: return (response_ok, notifications) - def repos_issues_search(self, username): + 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): - data = { - } + params = dict() - endpoint = self._Gitea__api_base_url + f'/api/v1/repos/issues/search?state=open&owner={username}&token={self._Gitea__access_token}' + 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) - response = self.__api_request('GET', endpoint, data) + endpoint = self.api_base_url + '/api/v1/repos/issues/search?{0}'.format(formatted_params) + + response = self.__api_request('GET', endpoint) + + response_ok = response.ok + + issues = response.json() + + return (response_ok, issues) + + 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): + """ + 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_ok = response.ok @@ -95,37 +174,28 @@ class Gitea: return (response_ok, notifications) - def repos_owner_repo_issues(self, username, repo): + 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) - data = { - } + endpoint = self.api_base_url + '/api/v1/repos/{0}/{1}/issues/comments?{2}'.format(owner, repo, formatted_params) - endpoint = self._Gitea__api_base_url + f'/api/v1/repos/{username}/{repo}/issues?state=all&token={self._Gitea__access_token}' - - response = self.__api_request('GET', endpoint, data) + response = self.__api_request('GET', endpoint)#, data) response_ok = response.ok - notifications = response.json() - - return (response_ok, notifications) - - def repos_owner_repo_issues_comments(self, username, repo): - - data = { - } - - endpoint = self._Gitea__api_base_url + f'/api/v1/repos/{username}/{repo}/issues/comments?token={self._Gitea__access_token}' - - response = self.__api_request('GET', endpoint, data) - - response_ok = response.ok - - notifications = response.json() - - return (response_ok, notifications) - + comments = response.json() + return (response_ok, comments) @staticmethod def __check_setup(self): @@ -145,8 +215,8 @@ class Gitea: 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: ") + self.api_base_url = input("Gitea API base url, in ex. 'https://yourgitea.instance': ") + self.access_token = input("Gitea access token: ") if not os.path.exists(self.__gitea_config_path): with open(self.__gitea_config_path, 'w'): pass @@ -154,9 +224,9 @@ class Gitea: 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') + the_file.write(f'api_base_url: {self.api_base_url}\n'+f'access_token: {self.access_token}\n') - return (api_base_url, access_token) + return (self.api_base_url, self.access_token) @staticmethod def __get_parameter(parameter, file_path ): @@ -169,13 +239,15 @@ class Gitea: print(f'{file_path} Missing parameter {parameter}') sys.exit(0) - def __api_request(self, method, endpoint, data): + def __api_request(self, method, endpoint, data={}): response = None try: - response = self.session.request(method, url = endpoint, data = data) + kwargs = dict(data=data) + + response = self.session.request(method, url = endpoint, **kwargs) except Exception as e: