Use **kwargs to pass request data if needed and added all endpoint params
This commit is contained in:
pare
be546301b6
commit
7443bc9e83
S'han modificat 1 arxius amb 128 adicions i 56 eliminacions
184
Gitea.py
184
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:
|
||||
|
||||
|
|
|
|||
Loading…
Referencia en una nova incidència