pull/1/head
fkarg 2019-04-11 22:21:26 +02:00
rodzic 938d187e6b
commit ec0f7a7259
1 zmienionych plików z 302 dodań i 210 usunięć

Wyświetl plik

@ -2,18 +2,20 @@ import json
import requests import requests
import logging import logging
logging = logging.getLogger('gitea') logging = logging.getLogger("gitea")
class AlreadyExistsException(Exception): class AlreadyExistsException(Exception):
""" Something (User/Repo/Organization/...) already exists. """ Something (User/Repo/Organization/...) already exists.
""" """
pass pass
class NotFoundException(Exception): class NotFoundException(Exception):
""" Something (User/Repo/Organization/...) has not been found. """ Something (User/Repo/Organization/...) has not been found.
""" """
pass pass
@ -70,10 +72,13 @@ class Organization:
Returns: [Repository] Returns: [Repository]
A list of Repositories this Organization is hosting. A list of Repositories this Organization is hosting.
""" """
results = self.gitea.requests_get(Organization.ORG_REPOS_REQUEST % results = self.gitea.requests_get(
self.username) Organization.ORG_REPOS_REQUEST % self.username
return [Repository(self.gitea, self, result["name"], )
initJson=result) for result in results] return [
Repository(self.gitea, self, result["name"], initJson=result)
for result in results
]
def get_teams(self): def get_teams(self):
""" Get the Teams in this Organization """ Get the Teams in this Organization
@ -81,10 +86,10 @@ class Organization:
Returns: [Team] Returns: [Team]
A list of Teams in this Organization. A list of Teams in this Organization.
""" """
results = self.gitea.requests_get(Organization.ORG_TEAMS_REQUEST % results = self.gitea.requests_get(
self.username) Organization.ORG_TEAMS_REQUEST % self.username
return [Team(self, result["name"], initJson=result) )
for result in results] return [Team(self, result["name"], initJson=result) for result in results]
def get_members(self): def get_members(self):
""" Get all members of this Organization """ Get all members of this Organization
@ -92,10 +97,8 @@ class Organization:
Returns: [User] Returns: [User]
A list of Users who are members of this Organization. A list of Users who are members of this Organization.
""" """
results = self.gitea.requests_get(Organization.ORG_GET_MEMBERS % results = self.gitea.requests_get(Organization.ORG_GET_MEMBERS % self.username)
self.username) return [User(self, result["username"], initJson=result) for result in results]
return [User(self, result["username"], initJson=result)
for result in results]
def __initialize_org(self, orgName: str, result) -> None: def __initialize_org(self, orgName: str, result) -> None:
""" Initialize Organization. """ Initialize Organization.
@ -107,15 +110,11 @@ class Organization:
Throws: Throws:
Exception, if Organization could not be found. Exception, if Organization could not be found.
""" """
try: if not result:
if not result: result = self.gitea.requests_get(Organization.ORG_REQUEST % orgName)
result = self.gitea.requests_get(Organization.ORG_REQUEST % logging.debug("Found Organization: %s" % orgName)
orgName) for i, v in result.items():
logging.debug("Found Organization: %s" % orgName) setattr(self, i, v)
for i, v in result.items():
setattr(self, i, v)
except Exception:
logging.error("Did not find organisation: %s" % orgName)
def set_value(self, values: dict): def set_value(self, values: dict):
""" Setting a certain value for an Organization. """ Setting a certain value for an Organization.
@ -127,8 +126,9 @@ class Organization:
location: string location: string
website: string website: string
""" """
result = self.gitea.requests_patch(Organization.ORG_PATCH % result = self.gitea.requests_patch(
self.username, data=values) Organization.ORG_PATCH % self.username, data=values
)
self.__initialize_org(self.username, result) self.__initialize_org(self.username, result)
def delete(self): def delete(self):
@ -191,8 +191,7 @@ class User:
Returns: [Repository] Returns: [Repository]
A list of Repositories this user owns. A list of Repositories this user owns.
""" """
result = self.gitea.requests_get(User.USER_REPOS_REQUEST % result = self.gitea.requests_get(User.USER_REPOS_REQUEST % self.username)
self.username)
return [Repository(self.gitea, self, r["name"]) for r in result] return [Repository(self.gitea, self, r["name"]) for r in result]
def __initialize_user(self, userName: str, result) -> None: def __initialize_user(self, userName: str, result) -> None:
@ -214,10 +213,12 @@ class User:
prev = self.email prev = self.email
result = self.gitea.requests_get(User.USER_MAIL % self.login) result = self.gitea.requests_get(User.USER_MAIL % self.login)
for mail in result: for mail in result:
if mail['primary']: if mail["primary"]:
self.email = mail['email'] self.email = mail["email"]
break break
logging.info("User %s updated Mail: <%s> to <%s>" % (self.login, prev, self.email)) logging.info(
"User %s updated Mail: <%s> to <%s>" % (self.login, prev, self.email)
)
def set_value(self, email: str, values: dict): def set_value(self, email: str, values: dict):
""" Set certain values of this user. """ Set certain values of this user.
@ -232,8 +233,7 @@ class User:
""" """
# the request requires email to be set... # the request requires email to be set...
values["email"] = email values["email"] = email
result = self.gitea.requests_patch(User.USER_PATCH % self.username, result = self.gitea.requests_patch(User.USER_PATCH % self.username, data=values)
data=values)
self.__initialize_user(self.username, result) self.__initialize_user(self.username, result)
def delete(self): def delete(self):
@ -261,7 +261,7 @@ class Repository:
... ...
""" """
REPO_REQUEST = """/repos/%s/%s""" # <owner>, <reponame> REPO_REQUEST = """/repos/%s/%s""" # <owner>, <reponame>
REPO_SEARCH = """/repos/search/%s""" # <reponame> REPO_SEARCH = """/repos/search/%s""" # <reponame>
REPO_BRANCHES = """/repos/%s/%s/branches""" # <owner>, <reponame> REPO_BRANCHES = """/repos/%s/%s/branches""" # <owner>, <reponame>
REPO_DELETE = """/repos/%s/%s""" # <owner>, <reponame> REPO_DELETE = """/repos/%s/%s""" # <owner>, <reponame>
@ -297,10 +297,10 @@ class Repository:
NotFoundException, if Repository has not been found. NotFoundException, if Repository has not been found.
""" """
if not result: if not result:
result = self.gitea.requests_get(Repository.REPO_REQUEST % result = self.gitea.requests_get(
(repoOwner.username, repoName)) Repository.REPO_REQUEST % (repoOwner.username, repoName)
logging.debug("Found Repository: %s/%s" % )
(repoOwner.username, repoName)) logging.debug("Found Repository: %s/%s" % (repoOwner.username, repoName))
for i, v in result.items(): for i, v in result.items():
setattr(self, i, v) setattr(self, i, v)
self.owner = repoOwner self.owner = repoOwner
@ -316,8 +316,9 @@ class Repository:
Returns: [Branch] Returns: [Branch]
A list of Branches of this Repository. A list of Branches of this Repository.
""" """
results = self.gitea.requests_get(Repository.REPO_BRANCHES % results = self.gitea.requests_get(
self.owner.username, self.name) Repository.REPO_BRANCHES % self.owner.username, self.name
)
return [Branch(self, result["name"], result) for result in results] return [Branch(self, result["name"], result) for result in results]
def delete(self): def delete(self):
@ -326,7 +327,9 @@ class Repository:
Warning: Warning:
Invalidates this objects Data. Invalidates this objects Data.
""" """
self.gitea.requests_delete(Repository.REPO_DELETE % (self.owner.username, self.name)) self.gitea.requests_delete(
Repository.REPO_DELETE % (self.owner.username, self.name)
)
class Branch: class Branch:
@ -368,11 +371,13 @@ class Branch:
NotFoundException, if Branch could not be found. NotFoundException, if Branch could not be found.
""" """
if not result: if not result:
result = self.gitea.requests_get(Branch.REPO_BRANCH % result = self.gitea.requests_get(
(repository.owner.username, Branch.REPO_BRANCH % (repository.owner.username, repository.name, name)
repository.name, name)) )
logging.debug("Branch found: %s/%s/%s" % (repository.owner.username, logging.debug(
repository.name, name)) "Branch found: %s/%s/%s"
% (repository.owner.username, repository.name, name)
)
for i, v in result.items(): for i, v in result.items():
setattr(self, i, v) setattr(self, i, v)
self.repository = repository self.repository = repository
@ -386,6 +391,7 @@ class Team:
GET_TEAM = """/teams/%s""" # <id> GET_TEAM = """/teams/%s""" # <id>
ADD_USER = """/teams/%s/members/%s""" # <id, username to add> ADD_USER = """/teams/%s/members/%s""" # <id, username to add>
ADD_REPO = """/teams/%s/repos/%s/%s""" # <id, org, repo> ADD_REPO = """/teams/%s/repos/%s/%s""" # <id, org, repo>
TEAM_DELETE = """/teams/%s""" # <id>
def __init__(self, org: Organization, name: str, initJson: json = None): def __init__(self, org: Organization, name: str, initJson: json = None):
""" Initializes Team. """ Initializes Team.
@ -421,8 +427,7 @@ class Team:
result = self.gitea.requests_get(Team.GET_TEAM % team.id) result = self.gitea.requests_get(Team.GET_TEAM % team.id)
logging.debug("Team found: %s/%s" % (org.username, name)) logging.debug("Team found: %s/%s" % (org.username, name))
if not result: if not result:
logging.warning("Failed to find Team: %s/%s" % logging.warning("Failed to find Team: %s/%s" % (org.username, name))
(org.username, name))
raise NotFoundException() raise NotFoundException()
for i, v in result.items(): for i, v in result.items():
setattr(self, i, v) setattr(self, i, v)
@ -447,10 +452,17 @@ class Team:
Args: Args:
repo (Repository): Repository to be added. repo (Repository): Repository to be added.
""" """
self.gitea.requests_put(Team.ADD_REPO % (self.id, self.organization.username, repo.name)) self.gitea.requests_put(
Team.ADD_REPO % (self.id, self.organization.username, repo.name)
)
def delete(self):
""" Delete this Team.
"""
self.gitea.requests_delete(Team.TEAM_DELETE % self.id)
class Gitea(): class Gitea:
""" Has Gitea-authenticated session. Can Create Users/Organizations/Teams/... """ Has Gitea-authenticated session. Can Create Users/Organizations/Teams/...
Attr: Attr:
@ -471,8 +483,10 @@ class Gitea():
url (str): URL of Gitea-server. url (str): URL of Gitea-server.
token (str): Token of acting User. token (str): Token of acting User.
""" """
self.headers = {"Authorization": "token " + token, self.headers = {
"Content-type": "application/json"} "Authorization": "token " + token,
"Content-type": "application/json",
}
self.url = url self.url = url
self.requests = requests self.requests = requests
@ -486,7 +500,7 @@ class Gitea():
Combined total API endpoint. Combined total API endpoint.
""" """
url = self.url + "/api/v1" + endpoint url = self.url + "/api/v1" + endpoint
logging.debug('Url: %s' % url) logging.debug("Url: %s" % url)
return url return url
@staticmethod @staticmethod
@ -499,11 +513,10 @@ class Gitea():
Returns: dict Returns: dict
Parsed from JSON Parsed from JSON
""" """
if (result.text and len(result.text) > 3): if result.text and len(result.text) > 3:
return json.loads(result.text) return json.loads(result.text)
return {} return {}
def requests_get(self, endpoint): def requests_get(self, endpoint):
""" Get parsed result from API-endpoint. """ Get parsed result from API-endpoint.
@ -516,15 +529,16 @@ class Gitea():
Throws: Throws:
Exception, if answer status code is not ok. Exception, if answer status code is not ok.
""" """
request = self.requests.get(self.get_url(endpoint), request = self.requests.get(self.get_url(endpoint), headers=self.headers)
headers=self.headers)
if request.status_code not in [200, 201]: if request.status_code not in [200, 201]:
if request.status_code in [404]: if request.status_code in [404]:
raise NotFoundException() raise NotFoundException()
logging.error("Received status code: %s (%s)" % logging.error(
(request.status_code, request.url)) "Received status code: %s (%s)" % (request.status_code, request.url)
raise Exception("Received status code: %s (%s)" % )
(request.status_code, request.url)) raise Exception(
"Received status code: %s (%s)" % (request.status_code, request.url)
)
return self.parse_result(request) return self.parse_result(request)
def requests_put(self, endpoint): def requests_put(self, endpoint):
@ -536,14 +550,16 @@ class Gitea():
Throws: Throws:
Exception, if answer status code is not ok. Exception, if answer status code is not ok.
""" """
request = self.requests.put(self.get_url(endpoint), request = self.requests.put(self.get_url(endpoint), headers=self.headers)
headers=self.headers)
if request.status_code not in [204]: if request.status_code not in [204]:
logging.error("Received status code: %s (%s) %s" % logging.error(
(request.status_code, request.url, request.text)) "Received status code: %s (%s) %s"
raise Exception("Received status code: %s (%s) %s" % % (request.status_code, request.url, request.text)
(request.status_code, request.url, )
request.text)) raise Exception(
"Received status code: %s (%s) %s"
% (request.status_code, request.url, request.text)
)
def requests_delete(self, endpoint): def requests_delete(self, endpoint):
""" Get parsed result from API-endpoint. """ Get parsed result from API-endpoint.
@ -554,13 +570,15 @@ class Gitea():
Throws: Throws:
Exception, if answer status code is not ok. Exception, if answer status code is not ok.
""" """
request = self.requests.delete(self.get_url(endpoint), request = self.requests.delete(self.get_url(endpoint), headers=self.headers)
headers=self.headers)
if request.status_code not in [204]: if request.status_code not in [204]:
logging.error("Received status code: %s (%s)" % logging.error(
(request.status_code, request.url)) "Received status code: %s (%s)" % (request.status_code, request.url)
raise Exception("Received status code: %s (%s) %s" % )
(request.status_code, request.url, vars(request))) raise Exception(
"Received status code: %s (%s) %s"
% (request.status_code, request.url, vars(request))
)
def requests_post(self, endpoint, data): def requests_post(self, endpoint, data):
""" Post data to API-endpoint. """ Post data to API-endpoint.
@ -576,22 +594,25 @@ class Gitea():
AlreadyExistsException, if 'already exists' in answer AlreadyExistsException, if 'already exists' in answer
Exception, if status code not ok Exception, if status code not ok
""" """
request = self.requests.post(self.get_url(endpoint), request = self.requests.post(
headers=self.headers, self.get_url(endpoint), headers=self.headers, data=json.dumps(data)
data=json.dumps(data)) )
if request.status_code not in [200, 201]: if request.status_code not in [200, 201]:
if 'already exists' in request.text or 'e-mail already in use' in request.text: if (
"already exists" in request.text
or "e-mail already in use" in request.text
):
logging.warning(request.text) logging.warning(request.text)
raise AlreadyExistsException() raise AlreadyExistsException()
logging.error("Received status code: %s (%s)" % logging.error(
(request.status_code, request.url)) "Received status code: %s (%s)" % (request.status_code, request.url)
logging.error("With info: %s (%s)" % )
(data, self.headers)) logging.error("With info: %s (%s)" % (data, self.headers))
logging.error("Answer: %s" % logging.error("Answer: %s" % request.text)
request.text) raise Exception(
raise Exception("Received status code: %s (%s), %s" % "Received status code: %s (%s), %s"
(request.status_code, request.url, % (request.status_code, request.url, request.text)
request.text)) )
return self.parse_result(request) return self.parse_result(request)
@ -608,116 +629,146 @@ class Gitea():
Throws: Throws:
Exception, if status code not ok. Exception, if status code not ok.
""" """
request = self.requests.patch(self.get_url(endpoint), request = self.requests.patch(
headers=self.headers, data=json.dumps(data)) self.get_url(endpoint), headers=self.headers, data=json.dumps(data)
)
if request.status_code not in [200, 201]: if request.status_code not in [200, 201]:
logging.error("Received status code: %s (%s) %s" % logging.error(
(request.status_code, request.url, data)) "Received status code: %s (%s) %s"
raise Exception("Received status code: %s (%s) %s" % % (request.status_code, request.url, data)
(request.status_code, request.url, request.text)) )
raise Exception(
"Received status code: %s (%s) %s"
% (request.status_code, request.url, request.text)
)
return self.parse_result(request) return self.parse_result(request)
def get_users_search(self): def get_users_search(self):
path = '/users/search' path = "/users/search"
return self.requests_get(path) return self.requests_get(path)
def delete_repos(self, username, reponame): def delete_repos(self, username, reponame):
path = '/repos/' + username + '/' + reponame path = "/repos/" + username + "/" + reponame
return self.requests.delete(path) return self.requests.delete(path)
def get_orgs_public_members_all(self, orgname): def get_orgs_public_members_all(self, orgname):
path = '/orgs/' + orgname + '/public_members' path = "/orgs/" + orgname + "/public_members"
return self.requests_get(path) return self.requests_get(path)
def post_repos__forks(self, organization, repo, owner): def post_repos__forks(self, organization, repo, owner):
path = '/repos/' + owner + '/' + repo + '/forks' path = "/repos/" + owner + "/" + repo + "/forks"
return self.requests_post(path, data={'organization': organization}) return self.requests_post(path, data={"organization": organization})
def get_repos_forks(self, repo, owner): def get_repos_forks(self, repo, owner):
path = '/repos/' + owner + '/' + repo + '/forks' path = "/repos/" + owner + "/" + repo + "/forks"
return self.requests_get(path) return self.requests_get(path)
def put_repos__subscription(self, username, reponame): def put_repos__subscription(self, username, reponame):
path = '/repos/' + username + '/' + reponame + '/subscription' path = "/repos/" + username + "/" + reponame + "/subscription"
return self.requests.put(path) return self.requests.put(path)
def delete_repos_subscription(self, username, reponame): def delete_repos_subscription(self, username, reponame):
path = '/repos/' + username + '/' + reponame + '/subscription' path = "/repos/" + username + "/" + reponame + "/subscription"
return self.requests.delete(path) return self.requests.delete(path)
def get_repos_subscription(self, username, reponame): def get_repos_subscription(self, username, reponame):
path = '/repos/' + username + '/' + reponame + '/subscription' path = "/repos/" + username + "/" + reponame + "/subscription"
return self.requests_get(path) return self.requests_get(path)
def get_users_following(self, username): def get_users_following(self, username):
path = '/users/' + username + '/following' path = "/users/" + username + "/following"
return self.requests_get(path) return self.requests_get(path)
def get_users_starred(self, username): def get_users_starred(self, username):
path = '/users/' + username + '/starred' path = "/users/" + username + "/starred"
return self.requests_get(path) return self.requests_get(path)
def put_orgs_public_members(self, username, orgname): def put_orgs_public_members(self, username, orgname):
path = '/orgs/' + orgname + '/public_members/' + username path = "/orgs/" + orgname + "/public_members/" + username
return self.requests.put(path) return self.requests.put(path)
def delete_orgs_public_members(self, username, orgname): def delete_orgs_public_members(self, username, orgname):
path = '/orgs/' + orgname + '/public_members/' + username path = "/orgs/" + orgname + "/public_members/" + username
return self.requests.delete(path) return self.requests.delete(path)
def get_orgs_public_members(self, username, orgname): def get_orgs_public_members(self, username, orgname):
path = '/orgs/' + orgname + '/public_members/' + username path = "/orgs/" + orgname + "/public_members/" + username
return self.requests_get(path) return self.requests_get(path)
def post_org_repos(self, name, description, private, auto_init, gitignores, def post_org_repos(
license, readme, org): self, name, description, private, auto_init, gitignores, license, readme, org
path = '/org/' + org + '/repos' ):
return self.requests_post(path, data={'name': name, path = "/org/" + org + "/repos"
'description': description, return self.requests_post(
'private': private, path,
'auto_init': auto_init, data={
'gitignores': gitignores, "name": name,
'license': license, "description": description,
'readme': readme}) "private": private,
"auto_init": auto_init,
"gitignores": gitignores,
"license": license,
"readme": readme,
},
)
def delete_orgs_members(self, orgname, username): def delete_orgs_members(self, orgname, username):
path = '/orgs/' + orgname + '/members/' + username path = "/orgs/" + orgname + "/members/" + username
return self.requests.delete(path) return self.requests.delete(path)
def post_repos__hooks(self, type, config, events, active, def post_repos__hooks(self, type, config, events, active, reponame, username):
reponame, username): path = "/repos/" + username + "/" + reponame + "/hooks"
path = '/repos/' + username + '/' + reponame + '/hooks' return self.requests_post(
return self.requests_post(path, data={'type': type, 'config': config, path,
'events': events, data={"type": type, "config": config, "events": events, "active": active},
'active': active}) )
def get_repos_hooks(self, reponame, username): def get_repos_hooks(self, reponame, username):
path = '/repos/' + username + '/' + reponame + '/hooks' path = "/repos/" + username + "/" + reponame + "/hooks"
return self.requests_get(path) return self.requests_get(path)
def post_repos_migrate(self, clone_addr, auth_username, auth_password, uid, def post_repos_migrate(
repo_name, mirror, private, self,
description): clone_addr,
path = '/repos/migrate' auth_username,
return self.requests_post(path, data={'clone_addr': clone_addr, auth_password,
'auth_username': auth_username, uid,
'auth_password': auth_password, repo_name,
'uid': uid, mirror,
'repo_name': repo_name, private,
'mirror': mirror, description,
'private': private, ):
'description': description}) path = "/repos/migrate"
return self.requests_post(
path,
data={
"clone_addr": clone_addr,
"auth_username": auth_username,
"auth_password": auth_password,
"uid": uid,
"repo_name": repo_name,
"mirror": mirror,
"private": private,
"description": description,
},
)
def post_user_repos(self, name, description, private, auto_init, def post_user_repos(
gitignores, license, readme): self, name, description, private, auto_init, gitignores, license, readme
path = '/user/repos' ):
return self.requests_post(path, data={'name': name, path = "/user/repos"
'description': description, return self.requests_post(
'private': private, path,
'auto_init': auto_init, data={
'gitignores': gitignores, "name": name,
'license': license, "description": description,
'readme': readme}) "private": private,
"auto_init": auto_init,
"gitignores": gitignores,
"license": license,
"readme": readme,
},
)
# # # # # #
@ -729,9 +780,15 @@ class Gitea():
result = self.requests_get(Gitea.GITEA_VERSION) result = self.requests_get(Gitea.GITEA_VERSION)
return result["version"] return result["version"]
def create_user(self, userName: str, email: str, password: str, def create_user(
change_pw=True, sendNotify=True, sourceId=0) \ self,
-> User: userName: str,
email: str,
password: str,
change_pw=True,
sendNotify=True,
sourceId=0,
) -> User:
""" Create User. """ Create User.
Args: Args:
@ -749,27 +806,40 @@ class Gitea():
AlreadyExistsException, if the User exists already AlreadyExistsException, if the User exists already
Exception, if something else went wrong. Exception, if something else went wrong.
""" """
result = self.requests_post(Gitea.ADMIN_CREATE_USER, result = self.requests_post(
data={'source_id': sourceId, Gitea.ADMIN_CREATE_USER,
'login_name': userName, data={
'username': userName, "source_id": sourceId,
'email': email, "login_name": userName,
'password': password, "username": userName,
'send_notify': sendNotify, "email": email,
'must_change_password': change_pw}) "password": password,
"send_notify": sendNotify,
"must_change_password": change_pw,
},
)
if "id" in result: if "id" in result:
logging.info("Successfully created User %s <%s> (id %s)" % logging.info(
(result["login"], result["email"], result["id"])) "Successfully created User %s <%s> (id %s)"
% (result["login"], result["email"], result["id"])
)
else: else:
logging.error(result["message"]) logging.error(result["message"])
raise Exception("User not created... (gitea: %s)" % raise Exception("User not created... (gitea: %s)" % result["message"])
result["message"])
return User(self, userName, result) return User(self, userName, result)
def create_repo(self, repoOwner, repoName: str, description: str='', def create_repo(
# private: bool=False, autoInit=True, gitignores='C#', self,
private: bool=False, autoInit=True, gitignores=None, repoOwner,
license=None, readme="Default") -> Repository: repoName: str,
description: str = "",
# private: bool=False, autoInit=True, gitignores='C#',
private: bool = False,
autoInit=True,
gitignores=None,
license=None,
readme="Default",
) -> Repository:
""" Create a Repository. """ Create a Repository.
Args: Args:
@ -792,28 +862,35 @@ class Gitea():
""" """
# although this only says user in the api, this also works for # although this only says user in the api, this also works for
# organizations # organizations
assert(isinstance(repoOwner, User) or assert isinstance(repoOwner, User) or isinstance(repoOwner, Organization)
isinstance(repoOwner, Organization)) result = self.requests_post(
result = self.requests_post(Gitea.ADMIN_REPO_CREATE % Gitea.ADMIN_REPO_CREATE % repoOwner.username,
repoOwner.username, data={
data={'name': repoName, "name": repoName,
'description': description, "description": description,
'private': private, "private": private,
'auto_init': autoInit, "auto_init": autoInit,
'gitignores': gitignores, "gitignores": gitignores,
'license': license, "license": license,
'readme': readme}) "readme": readme,
},
)
if "id" in result: if "id" in result:
logging.info("Successfully created Repository %s " % logging.info("Successfully created Repository %s " % result["name"])
result["name"])
else: else:
logging.error(result["message"]) logging.error(result["message"])
raise Exception("Repository not created... (gitea: %s)" % raise Exception("Repository not created... (gitea: %s)" % result["message"])
result["message"])
return Repository(self, repoOwner, repoName, result) return Repository(self, repoOwner, repoName, result)
def create_org(self, owner: User, orgName: str, description: str, def create_org(
location="", website="", full_name="") -> Organization: self,
owner: User,
orgName: str,
description: str,
location="",
website="",
full_name="",
) -> Organization:
""" Creates Organization. """ Creates Organization.
Args: Args:
@ -831,29 +908,43 @@ class Gitea():
AlreadyExistsException, if this Organization already exists. AlreadyExistsException, if this Organization already exists.
Exception, if something else went wrong. Exception, if something else went wrong.
""" """
assert (isinstance(owner, User)) assert isinstance(owner, User)
result = self.requests_post(Gitea.CREATE_ORG % owner.username, result = self.requests_post(
data={"username": orgName, Gitea.CREATE_ORG % owner.username,
"description": description, data={
"location": location, "username": orgName,
"website": website, "description": description,
"full_name": full_name}) "location": location,
"website": website,
"full_name": full_name,
},
)
if "id" in result: if "id" in result:
logging.info("Successfully created Organization %s" % logging.info("Successfully created Organization %s" % result["username"])
result["username"])
else: else:
logging.error("Organization not created... (gitea: %s)" % logging.error("Organization not created... (gitea: %s)" % result["message"])
result["message"])
logging.error(result["message"]) logging.error(result["message"])
raise Exception("Organization not created... (gitea: %s)" % raise Exception(
result["message"]) "Organization not created... (gitea: %s)" % result["message"]
)
return Organization(owner, orgName, initJson=result) return Organization(owner, orgName, initJson=result)
def create_team(self, org: Organization, name: str, description: str = '', def create_team(
permission: str = "read", self,
units = ["repo.code", "repo.issues", "repo.ext_issues", org: Organization,
"repo.wiki", "repo.pulls", "repo.releases", name: str,
"repo.ext_wiki"]) -> Team: description: str = "",
permission: str = "read",
units=[
"repo.code",
"repo.issues",
"repo.ext_issues",
"repo.wiki",
"repo.pulls",
"repo.releases",
"repo.ext_wiki",
],
) -> Team:
""" Creates a Team. """ Creates a Team.
Args: Args:
@ -862,18 +953,19 @@ class Gitea():
description (str): Optional, None, short description of the new Team. description (str): Optional, None, short description of the new Team.
permission (str): Optional, 'read', What permissions the members permission (str): Optional, 'read', What permissions the members
""" """
result = self.requests_post(Gitea.CREATE_TEAM % org.username, result = self.requests_post(
data={"name": name, Gitea.CREATE_TEAM % org.username,
"description": description, data={
"permission": permission, "name": name,
"units": units}) "description": description,
"permission": permission,
"units": units,
},
)
if "id" in result: if "id" in result:
logging.info("Successfully created Team %s" % logging.info("Successfully created Team %s" % result["name"])
result["name"])
else: else:
logging.error("Team not created... (gitea: %s)" % logging.error("Team not created... (gitea: %s)" % result["message"])
result["message"])
logging.error(result["message"]) logging.error(result["message"])
raise Exception("Team not created... (gitea: %s)" % raise Exception("Team not created... (gitea: %s)" % result["message"])
result["message"])
return Team(org, name, initJson=result) return Team(org, name, initJson=result)