mirror
/
py-gitea
kopia lustrzana https://github.com/Langenfeld/py-gitea
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

rename parse_request to parse_response, add function to repo

pull/3/head
danieldietsch 2019-12-17 17:52:06 +01:00
rodzic 4f05722a62
commit a363fae2ab
3 zmienionych plików z 47 dodań i 30 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

2
gitea/basicGiteaApiObject.py
Wyświetl plik

@ -32,7 +32,7 @@ class BasicGiteaApiObject:
return {name: getattr(self,name) for name in self.dirty_fields} return {name: getattr(self,name) for name in self.dirty_fields}
@classmethod @classmethod
def parse_request(cls, gitea, result): def parse_response(cls, gitea, result):
id = int(result["id"]) id = int(result["id"])
#gitea.logger.debug("Found api object of type %s (id: %s)" % (type(cls), id)) #gitea.logger.debug("Found api object of type %s (id: %s)" % (type(cls), id))
api_object = cls(gitea, id=id) api_object = cls(gitea, id=id)

73
gitea/gitea.py
Wyświetl plik

@ -1,7 +1,7 @@
import json import json
import logging import logging
from datetime import datetime from datetime import datetime
from typing import List, Tuple, Dict from typing import List, Tuple, Dict, Sequence
import requests import requests
from httpcache import CachingHTTPAdapter from httpcache import CachingHTTPAdapter
@ -39,11 +39,11 @@ class Organization(GiteaApiObject):
def get_repositories(self) -> List[GiteaApiObject]: def get_repositories(self) -> List[GiteaApiObject]:
results = self.gitea.requests_get(Organization.ORG_REPOS_REQUEST % self.username) results = self.gitea.requests_get(Organization.ORG_REPOS_REQUEST % self.username)
return [Repository.parse_request(self.gitea, result) for result in results] return [Repository.parse_response(self.gitea, result) for result in results]
def get_teams(self) -> List[GiteaApiObject]: def get_teams(self) -> List[GiteaApiObject]:
results = self.gitea.requests_get(Organization.ORG_TEAMS_REQUEST % self.username) results = self.gitea.requests_get(Organization.ORG_TEAMS_REQUEST % self.username)
return [Team.parse_request(self.gitea, result) for result in results] return [Team.parse_response(self.gitea, result) for result in results]
def get_team(self, name): def get_team(self, name):
teams = self.get_teams() teams = self.get_teams()
@ -54,7 +54,7 @@ class Organization(GiteaApiObject):
def get_members(self) -> List[GiteaApiObject]: def get_members(self) -> List[GiteaApiObject]:
results = self.gitea.requests_get(Organization.ORG_GET_MEMBERS % self.username) results = self.gitea.requests_get(Organization.ORG_GET_MEMBERS % self.username)
return [User.parse_request(self.gitea, result) for result in results] return [User.parse_response(self.gitea, result) for result in results]
def is_member(self, username) -> bool: def is_member(self, username) -> bool:
if isinstance(username, User): if isinstance(username, User):
@ -120,12 +120,12 @@ class User(GiteaApiObject):
def get_accessible_repositories(self) -> List[GiteaApiObject]: def get_accessible_repositories(self) -> List[GiteaApiObject]:
""" Get all Repositories owned by this User.""" """ Get all Repositories owned by this User."""
results = self.gitea.requests_get("/user/repos") results = self.gitea.requests_get("/user/repos")
return [Repository.parse_request(self.gitea, result) for result in results] return [Repository.parse_response(self.gitea, result) for result in results]
def get_repositories(self) -> List[GiteaApiObject]: def get_repositories(self) -> List[GiteaApiObject]:
""" Get all Repositories owned by this User.""" """ Get all Repositories owned by this User."""
results = self.gitea.requests_get(User.USER_REPOS_REQUEST % self.username) results = self.gitea.requests_get(User.USER_REPOS_REQUEST % self.username)
return [Repository.parse_request(self.gitea, result) for result in results] return [Repository.parse_response(self.gitea, result) for result in results]
def __request_emails(self): def __request_emails(self):
result = self.gitea.requests_get(User.USER_MAIL % self.login) result = self.gitea.requests_get(User.USER_MAIL % self.login)
@ -154,13 +154,14 @@ class Repository(GiteaApiObject):
REPO_ISSUES = """/repos/%s/%s/issues""" # <owner, reponame> REPO_ISSUES = """/repos/%s/%s/issues""" # <owner, reponame>
REPO_DELETE = """/repos/%s/%s""" # <owner>, <reponame> REPO_DELETE = """/repos/%s/%s""" # <owner>, <reponame>
REPO_USER_TIME = """/repos/%s/%s/times/%s""" # <owner>, <reponame>, <username> REPO_USER_TIME = """/repos/%s/%s/times/%s""" # <owner>, <reponame>, <username>
REPO_GET_COLLABORATOR = "/repos/{owner}/{repo}/collaborators" # <owner>, <reponame>
def __init__(self, gitea, id: int): def __init__(self, gitea, id: int):
super(Repository, self).__init__(gitea, id=id) super(Repository, self).__init__(gitea, id=id)
fields_to_parsers = { fields_to_parsers = {
# dont know how to tell apart user and org as owner except form email being empty. # dont know how to tell apart user and org as owner except form email being empty.
"owner": lambda gitea, r: Organization.parse_request(gitea, r) if r["email"] == "" else User.parse_request( "owner": lambda gitea, r: Organization.parse_response(gitea, r) if r["email"] == "" else User.parse_response(
gitea, r), gitea, r),
"updated_at": lambda gitea, t: Util.convert_time(t) "updated_at": lambda gitea, t: Util.convert_time(t)
} }
@ -176,7 +177,7 @@ class Repository(GiteaApiObject):
def get_branches(self) -> List[GiteaApiObject]: def get_branches(self) -> List[GiteaApiObject]:
"""Get all the Branches of this Repository.""" """Get all the Branches of this Repository."""
results = self.gitea.requests_get(Repository.REPO_BRANCHES % (self.owner.username, self.name)) results = self.gitea.requests_get(Repository.REPO_BRANCHES % (self.owner.username, self.name))
return [Branch.parse_request(self.gitea, result) for result in results] return [Branch.parse_response(self.gitea, result) for result in results]
def get_issues(self) -> List[GiteaApiObject]: def get_issues(self) -> List[GiteaApiObject]:
"""Get all Issues of this Repository (open and closed)""" """Get all Issues of this Repository (open and closed)"""
@ -190,7 +191,7 @@ class Repository(GiteaApiObject):
data = {"page": -1, "state": state} data = {"page": -1, "state": state}
results = self.gitea.requests_get(Repository.REPO_ISSUES % (self.owner.username, self.name), params=data) results = self.gitea.requests_get(Repository.REPO_ISSUES % (self.owner.username, self.name), params=data)
for result in results: for result in results:
issue = Issue.parse_request(self.gitea, result) issue = Issue.parse_response(self.gitea, result)
# adding data not contained in the issue answer # adding data not contained in the issue answer
setattr(issue, "repo", self.name) setattr(issue, "repo", self.name)
setattr(issue, "owner", self.owner) setattr(issue, "owner", self.owner)
@ -215,7 +216,7 @@ class Repository(GiteaApiObject):
"title": title "title": title
} }
result = self.gitea.requests_post(Repository.REPO_ISSUES % (self.owner.username, self.name), data=data) result = self.gitea.requests_post(Repository.REPO_ISSUES % (self.owner.username, self.name), data=data)
return Issue.parse_request(self.gitea, result) return Issue.parse_response(self.gitea, result)
def is_collaborator(self, username) -> bool: def is_collaborator(self, username) -> bool:
if isinstance(username, User): if isinstance(username, User):
@ -227,6 +228,21 @@ class Repository(GiteaApiObject):
except: except:
return False return False
def get_users_with_access(self) -> Sequence[User]:
response = self.gitea.requests_get(
Repository.REPO_GET_COLLABORATOR.format(owner=self.owner.username, repo=self.name))
collabs = [User.parse_response(self.gitea, user) for user in response]
if isinstance(self.owner, User):
return collabs + [self.owner]
else:
# owner must be org
teams = self.owner.get_teams()
for team in teams:
team_repos = team.get_repos()
if self.name in [n.name for n in team_repos]:
collabs += team.get_members()
return collabs
def delete(self): def delete(self):
self.gitea.requests_delete(Repository.REPO_DELETE % (self.owner.username, self.name)) self.gitea.requests_delete(Repository.REPO_DELETE % (self.owner.username, self.name))
self.deleted = True self.deleted = True
@ -259,7 +275,7 @@ class Comment(BasicGiteaApiObject):
super(Comment, self).__init__(gitea, id=id) super(Comment, self).__init__(gitea, id=id)
fields_to_parsers = { fields_to_parsers = {
"user": lambda gitea, r: User.parse_request(gitea, r), "user": lambda gitea, r: User.parse_response(gitea, r),
"created_at": lambda gitea, t: Util.convert_time(t), "created_at": lambda gitea, t: Util.convert_time(t),
"updated_at": lambda gitea, t: Util.convert_time(t) "updated_at": lambda gitea, t: Util.convert_time(t)
} }
@ -280,10 +296,10 @@ class Issue(GiteaApiObject):
super(Issue, self).__init__(gitea, id=id) super(Issue, self).__init__(gitea, id=id)
fields_to_parsers = { fields_to_parsers = {
"milestone": lambda gitea, m: Milestone.parse_request(gitea, m), "milestone": lambda gitea, m: Milestone.parse_response(gitea, m),
"user": lambda gitea, u: User.parse_request(gitea, u), "user": lambda gitea, u: User.parse_response(gitea, u),
"assignee": lambda gitea, u: User.parse_request(gitea, u), "assignee": lambda gitea, u: User.parse_response(gitea, u),
"assignees": lambda gitea, us: [User.parse_request(gitea, u) for u in us], "assignees": lambda gitea, us: [User.parse_response(gitea, u) for u in us],
"state": lambda gitea, s: Issue.CLOSED if s == "closed" else Issue.OPENED, "state": lambda gitea, s: Issue.CLOSED if s == "closed" else Issue.OPENED,
} }
@ -299,7 +315,7 @@ class Issue(GiteaApiObject):
args = {"owner": repo.owner.username, "repo": repo.name} args = {"owner": repo.owner.username, "repo": repo.name}
data = {"title": title, "body": body} data = {"title": title, "body": body}
result = gitea.requests_post(Issue.CREATE_ISSUE.format(**args), data=data) result = gitea.requests_post(Issue.CREATE_ISSUE.format(**args), data=data)
return Issue.parse_request(gitea, result) return Issue.parse_response(gitea, result)
def get_time(self, user: User) -> int: def get_time(self, user: User) -> int:
results = self.gitea.requests_get(Issue.GET_TIME % (self.owner.username, self.repo, self.number)) results = self.gitea.requests_get(Issue.GET_TIME % (self.owner.username, self.repo, self.number))
@ -307,7 +323,7 @@ class Issue(GiteaApiObject):
def get_comments(self) -> List[GiteaApiObject]: def get_comments(self) -> List[GiteaApiObject]:
results = self.gitea.requests_get(Issue.GET_COMMENTS % (self.owner.username, self.repo)) results = self.gitea.requests_get(Issue.GET_COMMENTS % (self.owner.username, self.repo))
allProjectComments = [Comment.parse_request(self.gitea, result) for result in results] allProjectComments = [Comment.parse_response(self.gitea, result) for result in results]
# Comparing the issue id with the URL seems to be the only (!) way to get to the comments of one issue # Comparing the issue id with the URL seems to be the only (!) way to get to the comments of one issue
return [comment for comment in allProjectComments if comment.issue_url.endswith("/" + str(self.number))] return [comment for comment in allProjectComments if comment.issue_url.endswith("/" + str(self.number))]
@ -335,7 +351,7 @@ class Team(GiteaApiObject):
super(Team, self).__init__(gitea, id=id) super(Team, self).__init__(gitea, id=id)
fields_to_parsers = { fields_to_parsers = {
"organization": lambda gitea, o: Organization.parse_request(gitea, o) "organization": lambda gitea, o: Organization.parse_response(gitea, o)
} }
@classmethod @classmethod
@ -353,12 +369,12 @@ class Team(GiteaApiObject):
def get_members(self): def get_members(self):
""" Get all users assigned to the team. """ """ Get all users assigned to the team. """
results = self.gitea.requests_get(Team.GET_MEMBERS % self.id) results = self.gitea.requests_get(Team.GET_MEMBERS % self.id)
return [User.parse_request(self.gitea, result) for result in results] return [User.parse_response(self.gitea, result) for result in results]
def get_repos(self): def get_repos(self):
""" Get all repos of this Team.""" """ Get all repos of this Team."""
results = self.gitea.requests_get(Team.GET_REPOS % self.id) results = self.gitea.requests_get(Team.GET_REPOS % self.id)
return [Repository.parse_request(self.gitea, result) for result in results] return [Repository.parse_response(self.gitea, result) for result in results]
def delete(self): def delete(self):
self.gitea.requests_delete(Team.TEAM_DELETE % self.id) self.gitea.requests_delete(Team.TEAM_DELETE % self.id)
@ -419,7 +435,8 @@ class Gitea:
if request.status_code in [404]: if request.status_code in [404]:
raise NotFoundException(message) raise NotFoundException(message)
if request.status_code in [403]: if request.status_code in [403]:
raise Exception("Unauthorized: %s - Check your permissions and try again! (%s)" % (request.url, message)) raise Exception(
"Unauthorized: %s - Check your permissions and try again! (%s)" % (request.url, message))
raise Exception(message) raise Exception(message)
return self.parse_result(request) return self.parse_result(request)
@ -488,7 +505,7 @@ class Gitea:
def get_orgs(self): def get_orgs(self):
path = "/admin/orgs" path = "/admin/orgs"
results = self.requests_get(path) results = self.requests_get(path)
return [Organization.parse_request(self, result) for result in results] return [Organization.parse_response(self, result) for result in results]
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"
@ -569,7 +586,7 @@ class Gitea:
def get_user(self): def get_user(self):
result = self.requests_get(Gitea.GET_USER) result = self.requests_get(Gitea.GET_USER)
return User.parse_request(self, result) return User.parse_response(self, result)
def get_version(self) -> str: def get_version(self) -> str:
result = self.requests_get(Gitea.GITEA_VERSION) result = self.requests_get(Gitea.GITEA_VERSION)
@ -577,7 +594,7 @@ class Gitea:
def get_users(self): def get_users(self):
results = self.requests_get(Gitea.GET_USERS_ADMIN) results = self.requests_get(Gitea.GET_USERS_ADMIN)
return [User.parse_request(self, result) for result in results] return [User.parse_response(self, result) for result in results]
def get_user_by_email(self, email: str) -> User: def get_user_by_email(self, email: str) -> User:
users = self.get_users() users = self.get_users()
@ -609,7 +626,7 @@ class Gitea:
else: else:
self.logger.error(result["message"]) self.logger.error(result["message"])
raise Exception("User not created... (gitea: %s)" % result["message"]) raise Exception("User not created... (gitea: %s)" % result["message"])
user = User.parse_request(self, result) user = User.parse_response(self, result)
return user return user
def create_repo(self, repoOwner, repoName: str, description: str = "", private: bool = False, autoInit=True, def create_repo(self, repoOwner, repoName: str, description: str = "", private: bool = False, autoInit=True,
@ -633,7 +650,7 @@ class Gitea:
else: else:
self.logger.error(result["message"]) self.logger.error(result["message"])
raise Exception("Repository not created... (gitea: %s)" % result["message"]) raise Exception("Repository not created... (gitea: %s)" % result["message"])
return Repository.parse_request(self, result) return Repository.parse_response(self, result)
def create_org(self, owner: User, orgName: str, description: str, location="", website="", full_name="", ): def create_org(self, owner: User, orgName: str, description: str, location="", website="", full_name="", ):
assert isinstance(owner, User) assert isinstance(owner, User)
@ -649,7 +666,7 @@ class Gitea:
raise Exception( raise Exception(
"Organization not created... (gitea: %s)" % result["message"] "Organization not created... (gitea: %s)" % result["message"]
) )
return Organization.parse_request(self, result) return Organization.parse_response(self, result)
def create_team(self, org: Organization, name: str, description: str = "", permission: str = "read", def create_team(self, org: Organization, name: str, description: str = "", permission: str = "read",
units=("repo.code", "repo.issues", "repo.ext_issues", "repo.wiki", "repo.pulls", "repo.releases", units=("repo.code", "repo.issues", "repo.ext_issues", "repo.wiki", "repo.pulls", "repo.releases",
@ -671,7 +688,7 @@ class Gitea:
self.logger.error("Team not created... (gitea: %s)" % result["message"]) self.logger.error("Team not created... (gitea: %s)" % result["message"])
self.logger.error(result["message"]) self.logger.error(result["message"])
raise Exception("Team not created... (gitea: %s)" % result["message"]) raise Exception("Team not created... (gitea: %s)" % result["message"])
api_object = Team.parse_request(self, result) api_object = Team.parse_response(self, result)
setattr(api_object, "_organization", setattr(api_object, "_organization",
org) # fixes strange behaviour of gitea not returning a valid organization here. org) # fixes strange behaviour of gitea not returning a valid organization here.
return api_object return api_object

2
gitea/giteaApiObject.py
Wyświetl plik

@ -18,7 +18,7 @@ class GiteaApiObject(BasicGiteaApiObject):
@classmethod @classmethod
def _request(cls, gitea, args): def _request(cls, gitea, args):
result = cls._get_gitea_api_object(gitea, args) result = cls._get_gitea_api_object(gitea, args)
api_object = cls.parse_request(gitea, result) api_object = cls.parse_response(gitea, result)
for key, value in args.items(): # hack: not all necessary request args in api result (e.g. repo name in issue) for key, value in args.items(): # hack: not all necessary request args in api result (e.g. repo name in issue)
if not hasattr(api_object, key): if not hasattr(api_object, key):
setattr(api_object, key, value) setattr(api_object, key, value)