From ad9dae0b0e91fe79c54467fa793e29ad133d4bf3 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Mon, 15 Nov 2021 13:47:43 +0100 Subject: [PATCH 01/13] cleaned up and renamed gitea api object bases --- ...asicGiteaApiObject.py => baseapiobject.py} | 129 +++++++++++------- gitea/exceptions.py | 5 + gitea/gitea.py | 38 +++--- gitea/giteaApiObject.py | 27 ---- 4 files changed, 102 insertions(+), 97 deletions(-) rename gitea/{basicGiteaApiObject.py => baseapiobject.py} (54%) delete mode 100644 gitea/giteaApiObject.py diff --git a/gitea/basicGiteaApiObject.py b/gitea/baseapiobject.py similarity index 54% rename from gitea/basicGiteaApiObject.py rename to gitea/baseapiobject.py index 9903604..89fca9a 100644 --- a/gitea/basicGiteaApiObject.py +++ b/gitea/baseapiobject.py @@ -1,14 +1,12 @@ -from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation +from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation,RawRequestEndpointMissing -class BasicGiteaApiObject: - GET_API_OBJECT = "FORMAT/STINING/{argument}" - PATCH_API_OBJECT = "FORMAT/STINING/{argument}" + +class ReadonlyApiObject: def __init__(self, gitea): self.gitea = gitea self.deleted = False # set if .delete was called, so that an exception is risen - self.dirty_fields = set() def __str__(self): return "GiteaAPIObject (%s):" % (type(self)) @@ -21,71 +19,100 @@ class BasicGiteaApiObject: """Hash only fields that are part of the gitea-data identity""" raise MissiongEqualyImplementation() - fields_to_parsers = {} + @classmethod + def request(cls, gitea, id): + if hasattr("GET_API_OBJECT", cls): + return cls._request(gitea) + else: + raise RawRequestEndpointMissing() + + @classmethod + def _request(cls, gitea, args): + result = cls._get_gitea_api_object(gitea, args) + api_object = cls.parse_response(gitea, result) + # hack: not all necessary request args in api result (e.g. repo name in issue) + for key, value in args.items(): + if not hasattr(api_object, key): + setattr(api_object, key, value) + return api_object + + @classmethod + def _get_gitea_api_object(cls, gitea, args): + """Retrieving an object always as GET_API_OBJECT """ + return gitea.requests_get(cls.GET_API_OBJECT.format(**args)) + + @classmethod + def parse_response(cls, gitea, result) -> "ReadonlyApiObject": + # gitea.logger.debug("Found api object of type %s (id: %s)" % (type(cls), id)) + api_object = cls(gitea) + cls._initialize(gitea, api_object, result) + return api_object + + @classmethod + def _initialize(cls, gitea, api_object, result): + for name, value in result.items(): + if name in cls.fields_to_parsers and value is not None: + parse_func = cls.fields_to_parsers[name] + value = parse_func(gitea, value) + cls._add_read_property(name, value, api_object) + # add all patchable fields missing in the request to be writable + for name in cls.fields_to_parsers.keys(): + if not hasattr(api_object,name): + cls._add_read_property(name, None, api_object) + + @classmethod + def _add_read_property(cls, name, value, api_object): + if not hasattr(api_object, name): + prop = property( + (lambda name: lambda self: self._get_var(name))(name)) + setattr(cls, name, prop) + setattr(api_object, "_" + name, value) + else: + raise AttributeError(f"Attribute {name} already exists on api object.") + + def _get_var(self, name): + if self.deleted: + raise ObjectIsInvalid() + return getattr(self, "_" + name) + + +class ApiObject(ReadonlyApiObject): + + patchable_fields = set() + + def __init__(self, gitea): + super().__init__(gitea) + self.dirty_fields = set() + def commit(self): raise NotImplemented() def get_dirty_fields(self): return {name: getattr(self, name) for name in self.dirty_fields} - @classmethod - def parse_response(cls, gitea, result) -> "BasicGiteaApiObject": - # gitea.logger.debug("Found api object of type %s (id: %s)" % (type(cls), id)) - api_object = cls(gitea) - cls._initialize(gitea, api_object, result) - return api_object - - @classmethod - def _get_gitea_api_object(cls, gitea, args): - """Retrieving an object always as GET_API_OBJECT """ - return gitea.requests_get(cls.GET_API_OBJECT.format(**args)) - - patchable_fields = set() - @classmethod def _initialize(cls, gitea, api_object, result): + super()._initialize(gitea,api_object,result) for name, value in result.items(): - if name in cls.fields_to_parsers and value is not None: - parse_func = cls.fields_to_parsers[name] - value = parse_func(gitea, value) if name in cls.patchable_fields: - cls._add_property(name, value, api_object) - else: - cls._add_readonly_property(name,value,api_object) + cls._add_write_property(name,value,api_object) # add all patchable fields missing in the request to be writable for name in cls.patchable_fields: if not hasattr(api_object,name): - cls._add_property(name, None, api_object) - for name in cls.fields_to_parsers.keys(): - if not hasattr(api_object,name): - cls._add_readonly_property(name, None, api_object) + cls._add_write_property(name, None, api_object) @classmethod - def _add_property(cls, name, value, api_object): - if not hasattr(api_object, name): - prop = property( - (lambda name: lambda self: self.__get_var(name))(name), - (lambda name: lambda self, v: self.__set_var(name, v))(name)) - setattr(cls, name, prop) - setattr(api_object, "_" + name, value) - - @classmethod - def _add_readonly_property(cls, name, value, api_object): - if not hasattr(api_object, name): - prop = property( - (lambda name: lambda self: self.__get_var(name))(name)) - setattr(cls, name, prop) - setattr(api_object, "_" + name, value) + def _add_write_property(cls, name, value, api_object): + prop = property( + (lambda name: lambda self: self._get_var(name))(name), + (lambda name: lambda self, v: self.__set_var(name, v))(name)) + setattr(cls, name, prop) + setattr(api_object, "_" + name, value) def __set_var(self, name, i): if self.deleted: raise ObjectIsInvalid() self.dirty_fields.add(name) - setattr(self, "_" + name, i) - - def __get_var(self, name): - if self.deleted: - raise ObjectIsInvalid() - return getattr(self, "_" + name) + setattr(self, "_" + name, i) \ No newline at end of file diff --git a/gitea/exceptions.py b/gitea/exceptions.py index 15ad497..f693ddb 100644 --- a/gitea/exceptions.py +++ b/gitea/exceptions.py @@ -13,6 +13,11 @@ class ObjectIsInvalid(Exception): class ConflictException(Exception): pass +class RawRequestEndpointMissing(Exception): + """This ApiObject can only be obtained through other api objects and does not have + diret .request method.""" + pass + class MissiongEqualyImplementation(Exception): """ Each Object obtained from the gitea api must be able to check itself for equality in relation to its diff --git a/gitea/gitea.py b/gitea/gitea.py index 4551749..037162c 100644 --- a/gitea/gitea.py +++ b/gitea/gitea.py @@ -6,12 +6,11 @@ from typing import List, Tuple, Dict, Sequence, Optional, Union, Set import requests from httpcache import CachingHTTPAdapter -from .basicGiteaApiObject import BasicGiteaApiObject +from .baseapiobject import ReadonlyApiObject, ApiObject from .exceptions import * -from .giteaApiObject import GiteaApiObject -class Organization(GiteaApiObject): +class Organization(ApiObject): """see https://try.gitea.io/api/swagger#/organization/orgGetAll""" GET_API_OBJECT = """/orgs/{name}""" # @@ -42,7 +41,7 @@ class Organization(GiteaApiObject): def parse_response(cls, gitea, result): api_object = super().parse_response(gitea, result) # add "name" field to make this behave similar to users - Organization._add_readonly_property("name", result["username"], api_object) + Organization._add_read_property("name", result["username"], api_object) return api_object patchable_fields = {"description", "full_name", "location", "visibility", "website"} @@ -100,7 +99,7 @@ class Organization(GiteaApiObject): except: return False - def remove_member(self, user: GiteaApiObject): + def remove_member(self, user: "User"): path = f"/orgs/{self.username}/members/{user.username}" self.gitea.requests_delete(path) @@ -120,7 +119,7 @@ class Organization(GiteaApiObject): return results -class User(GiteaApiObject): +class User(ApiObject): GET_API_OBJECT = """/users/{name}""" # USER_MAIL = """/user/emails?sudo=%s""" # USER_PATCH = """/admin/users/%s""" # @@ -226,18 +225,19 @@ class User(GiteaApiObject): return results -class Branch(GiteaApiObject): +class Branch(ReadonlyApiObject): GET_API_OBJECT = """/repos/%s/%s/branches/%s""" # , , def __init__(self, gitea): super(Branch, self).__init__(gitea) def __eq__(self, other): - if not isinstance(other, Branch): return False + if not isinstance(other, Branch): + return False return self.commit == other.commit and self.name == other.name def __hash__(self): - return hash(self.commit) ^ hash(self.name) + return hash(self.commit["id"]) ^ hash(self.name) fields_to_parsers = { # This is not a commit object @@ -249,7 +249,7 @@ class Branch(GiteaApiObject): return cls._request(gitea, {"owner": owner, "repo": repo, "ref": ref}) -class Repository(GiteaApiObject): +class Repository(ApiObject): REPO_IS_COLLABORATOR = ( """/repos/%s/%s/collaborators/%s""" # , , ) @@ -374,7 +374,7 @@ class Repository(GiteaApiObject): def get_full_name(self) -> str: return self.owner.username + "/" + self.name - def create_issue(self, title, assignees=[], description="") -> GiteaApiObject: + def create_issue(self, title, assignees=[], description="") -> ApiObject: data = { "assignees": assignees, "body": description, @@ -475,7 +475,7 @@ class Repository(GiteaApiObject): ) self.deleted = True -class Milestone(GiteaApiObject): +class Milestone(ApiObject): GET_API_OBJECT = ( """/repos/{owner}/{repo}/milestones/{number}""" # ) @@ -517,7 +517,7 @@ class Milestone(GiteaApiObject): return cls._request(gitea, {"owner": owner, "repo": repo, "number": number}) -class Comment(BasicGiteaApiObject): +class Comment(ApiObject): PATCH_API_OBJECT = "/repos/{owner}/{repo}/issues/comments/{id}" def __init__(self, gitea): @@ -539,7 +539,7 @@ class Comment(BasicGiteaApiObject): patchable_fields = {"body"} -class Commit(GiteaApiObject): +class Commit(ReadonlyApiObject): def __init__(self, gitea): super(Commit, self).__init__(gitea) @@ -565,11 +565,11 @@ class Commit(GiteaApiObject): commit_cache = result["commit"] api_object = cls(gitea) cls._initialize(gitea, api_object, result) - BasicGiteaApiObject._add_readonly_property("inner_commit", commit_cache, api_object) + Commit._add_read_property("inner_commit", commit_cache, api_object) return api_object -class Issue(GiteaApiObject): +class Issue(ApiObject): GET_API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # GET_TIME = """/repos/%s/%s/issues/%s/times""" # GET_COMMENTS = """/repos/%s/%s/issues/comments""" @@ -645,7 +645,7 @@ class Issue(GiteaApiObject): path, data={"created": created, "time": int(time), "user_name": user_name} ) - def get_comments(self) -> List[GiteaApiObject]: + def get_comments(self) -> List[ApiObject]: results = self.gitea.requests_get( Issue.GET_COMMENTS % (self.owner.username, self.repo) ) @@ -660,7 +660,7 @@ class Issue(GiteaApiObject): ] -class Team(GiteaApiObject): +class Team(ApiObject): GET_API_OBJECT = """/teams/{id}""" # ADD_USER = """/teams/%s/members/%s""" # ADD_REPO = """/teams/%s/repos/%s/%s""" # @@ -712,7 +712,7 @@ class Team(GiteaApiObject): url = f"/teams/{self.id}/members/{user_name}" self.gitea.requests_delete(url) -class Content(GiteaApiObject): +class Content(ReadonlyApiObject): GET_API_OBJECT = """/repos/{owner}/{repo}/contents/{filepath}""" FILE = "file" diff --git a/gitea/giteaApiObject.py b/gitea/giteaApiObject.py deleted file mode 100644 index 23e8479..0000000 --- a/gitea/giteaApiObject.py +++ /dev/null @@ -1,27 +0,0 @@ -from .basicGiteaApiObject import BasicGiteaApiObject - - -class GiteaApiObject(BasicGiteaApiObject): - GET_API_OBJECT = "FORMAT/STRING/{argument}" - PATCH_API_OBJECT = "FORMAT/STRING/{argument}" - - def __init__(self, gitea): - super(GiteaApiObject, self).__init__(gitea) - - @classmethod - def request(cls, gitea, id): - """Use for giving a nice e.g. 'request(gita, orgname, repo, ticket)'. - All args are put into an args tuple for passing around""" - return cls._request(gitea) - - @classmethod - def _request(cls, gitea, args): - result = cls._get_gitea_api_object(gitea, args) - api_object = cls.parse_response(gitea, result) - # hack: not all necessary request args in api result (e.g. repo name in issue) - for key, value in args.items(): - if not hasattr(api_object, key): - setattr(api_object, key, value) - return api_object - - From 29e581f7dde649ea4d8c7190f6b1057d07202939 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Mon, 15 Nov 2021 14:39:55 +0100 Subject: [PATCH 02/13] and further cleanup --- gitea/__init__.py | 12 +- gitea/apiobject.py | 716 +++++++++++++++++++++++++++++++++++++++ gitea/baseapiobject.py | 6 +- gitea/gitea.py | 737 +---------------------------------------- setup.py | 2 +- tests/test_api.py | 7 +- 6 files changed, 738 insertions(+), 742 deletions(-) create mode 100644 gitea/apiobject.py diff --git a/gitea/__init__.py b/gitea/__init__.py index 366c91c..97bbedd 100644 --- a/gitea/__init__.py +++ b/gitea/__init__.py @@ -1,5 +1,9 @@ from .gitea import ( Gitea, + NotFoundException, + AlreadyExistsException, +) +from .apiobject import ( User, Organization, Team, @@ -9,6 +13,9 @@ from .gitea import ( AlreadyExistsException, Issue, Milestone, + Commit, + Comment, + Content ) __all__ = [ @@ -21,5 +28,8 @@ __all__ = [ 'NotFoundException', 'AlreadyExistsException', 'Issue', - 'Milestone' + 'Milestone', + 'Commit', + 'Comment', + 'Content' ] diff --git a/gitea/apiobject.py b/gitea/apiobject.py new file mode 100644 index 0000000..38fc24f --- /dev/null +++ b/gitea/apiobject.py @@ -0,0 +1,716 @@ +import logging +from datetime import datetime +from typing import List, Tuple, Dict, Sequence, Optional, Union, Set + +from .baseapiobject import ReadonlyApiObject, ApiObject +from .exceptions import * + +class Organization(ApiObject): + """see https://try.gitea.io/api/swagger#/organization/orgGetAll""" + + API_OBJECT = """/orgs/{name}""" # + ORG_REPOS_REQUEST = """/orgs/%s/repos""" # + ORG_TEAMS_REQUEST = """/orgs/%s/teams""" # + ORG_TEAMS_CREATE = """/orgs/%s/teams""" # + ORG_GET_MEMBERS = """/orgs/%s/members""" # + ORG_IS_MEMBER = """/orgs/%s/members/%s""" # , + ORG_HEATMAP = """/users/%s/heatmap""" # + + def __init__(self, gitea): + super(Organization, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Organization): return False + return self.gitea == other.gitea and self.name == other.name + + def __hash__(self): + return hash(self.gitea) ^ hash(self.name) + + @classmethod + def request(cls, gitea, name): + return cls._request(gitea, {"name": name}) + + @classmethod + def parse_response(cls, gitea, result): + api_object = super().parse_response(gitea, result) + # add "name" field to make this behave similar to users + Organization._add_read_property("name", result["username"], api_object) + return api_object + + patchable_fields = {"description", "full_name", "location", "visibility", "website"} + + def commit(self): + values = self.get_dirty_fields() + args = {"name": self.name} + self.gitea.requests_patch( + Organization.API_OBJECT.format(**args), data=values + ) + self.dirty_fields = {} + + def get_repositories(self) -> List["Repository"]: + results = self.gitea.requests_get_paginated( + Organization.ORG_REPOS_REQUEST % self.username + ) + return [Repository.parse_response(self.gitea, result) for result in results] + + def get_repository(self, name) -> "Repository": + repos = self.get_repositories() + for repo in repos: + if repo.name == name: + return repo + raise NotFoundException("Repository %s not existent in organization." % name) + + def get_teams(self) -> List["Team"]: + results = self.gitea.requests_get( + Organization.ORG_TEAMS_REQUEST % self.username + ) + teams = [Team.parse_response(self.gitea, result) for result in results] + # organisation seems to be missing using this request, so we add org manually + for t in teams: setattr(t, "_organization", self) + return teams + + def get_team(self, name) -> "Team": + teams = self.get_teams() + for team in teams: + if team.name == name: + return team + raise NotFoundException("Team not existent in organization.") + + def get_members(self) -> List["User"]: + results = self.gitea.requests_get(Organization.ORG_GET_MEMBERS % self.username) + return [User.parse_response(self.gitea, result) for result in results] + + def is_member(self, username) -> bool: + if isinstance(username, User): + username = username.username + try: + # returns 204 if its ok, 404 if its not + self.gitea.requests_get( + Organization.ORG_IS_MEMBER % (self.username, username) + ) + return True + except: + return False + + def remove_member(self, user: "User"): + path = f"/orgs/{self.username}/members/{user.username}" + self.gitea.requests_delete(path) + + def delete(self): + """ Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User""" + for repo in self.get_repositories(): + repo.delete() + self.gitea.requests_delete(Organization.API_OBJECT.format(name=self.username)) + self.deleted = True + + def get_heatmap(self) -> List[Tuple[datetime, int]]: + results = self.gitea.requests_get(User.USER_HEATMAP % self.username) + results = [ + (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) + for result in results + ] + return results + + +class User(ApiObject): + API_OBJECT = """/users/{name}""" # + USER_MAIL = """/user/emails?sudo=%s""" # + USER_PATCH = """/admin/users/%s""" # + ADMIN_DELETE_USER = """/admin/users/%s""" # + ADMIN_EDIT_USER = """/admin/users/{username}""" # + USER_HEATMAP = """/users/%s/heatmap""" # + + def __init__(self, gitea): + super(User, self).__init__(gitea) + self._emails = [] + + def __eq__(self, other): + if not isinstance(other, User): return False + return self.gitea == other.gitea and self.id == other.id + + def __hash__(self): + return hash(self.gitea) ^ hash(self.id) + + @property + def emails(self): + self.__request_emails() + return self._emails + + @classmethod + def request(cls, gitea, name) -> "User": + api_object = cls._request(gitea, {"name": name}) + return api_object + + patchable_fields = { + "active", + "admin", + "allow_create_organization", + "allow_git_hook", + "allow_import_local", + "email", + "full_name", + "location", + "login_name", + "max_repo_creation", + "must_change_password", + "password", + "prohibit_login", + "website", + } + + def commit(self, login_name: str, source_id: int = 0): + """ + Unfortunately it is necessary to require the login name + as well as the login source (that is not supplied when getting a user) for + changing a user. + Usually source_id is 0 and the login_name is equal to the username. + """ + values = self.get_dirty_fields() + values.update( + # api-doc says that the "source_id" is necessary; works without though + {"login_name": login_name, "source_id": source_id} + ) + args = {"username": self.username} + self.gitea.requests_patch(User.ADMIN_EDIT_USER.format(**args), data=values) + self.dirty_fields = {} + + def get_repositories(self) -> List["Repository"]: + """ Get all Repositories owned by this User.""" + url = f"/users/{self.username}/repos" + results = self.gitea.requests_get(url) + return [Repository.parse_response(self.gitea, result) for result in results] + + def get_orgs(self) -> List[Organization]: + """ Get all Organizations this user is a member of.""" + url = f"/users/{self.username}/orgs" + results = self.gitea.requests_get(url) + return [Organization.parse_response(self.gitea, result) for result in results] + + def get_teams(self) -> List['Team']: + url = f"/user/teams" + results = self.gitea.requests_get(url, sudo=self) + return [Team.parse_response(self.gitea, result) for result in results] + + def get_accessible_repos(self) -> List['Repository']: + """ Get all Repositories accessible by the logged in User.""" + results = self.gitea.requests_get("/user/repos", sudo=self) + return [Repository.parse_response(self, result) for result in results] + + def __request_emails(self): + result = self.gitea.requests_get(User.USER_MAIL % self.login) + # report if the adress changed by this + for mail in result: + self._emails.append(mail["email"]) + if mail["primary"]: + self._email = mail["email"] + + def delete(self): + """ Deletes this User. Also deletes all Repositories he owns.""" + self.gitea.requests_delete(User.ADMIN_DELETE_USER % self.username) + self.deleted = True + + def get_heatmap(self) -> List[Tuple[datetime, int]]: + results = self.gitea.requests_get(User.USER_HEATMAP % self.username) + results = [ + (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) + for result in results + ] + return results + + +class Branch(ReadonlyApiObject): + + def __init__(self, gitea): + super(Branch, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Branch): + return False + return self.commit == other.commit and self.name == other.name + + def __hash__(self): + return hash(self.commit["id"]) ^ hash(self.name) + + fields_to_parsers = { + # This is not a commit object + #"commit": lambda gitea, c: Commit.parse_response(gitea, c) + } + + @classmethod + def request(cls, gitea, owner, repo, ref): + return cls._request(gitea, {"owner": owner, "repo": repo, "ref": ref}) + + +class Repository(ApiObject): + API_OBJECT = """/repos/{owner}/{name}""" # , + REPO_IS_COLLABORATOR = """/repos/%s/%s/collaborators/%s""" # , , + REPO_SEARCH = """/repos/search/%s""" # + REPO_BRANCHES = """/repos/%s/%s/branches""" # , + REPO_ISSUES = """/repos/%s/%s/issues""" # + REPO_DELETE = """/repos/%s/%s""" # , + REPO_TIMES = """/repos/%s/%s/times""" # , + REPO_USER_TIME = """/repos/%s/%s/times/%s""" # , , + REPO_COMMITS = "/repos/%s/%s/commits" # , + REPO_TRANSFER = "/repos/{owner}/{repo}/transfer" + REPO_CONTENTS = "/repos/{owner}/{repo}/contents" + REPO_CONTENT = """/repos/{owner}/{repo}/contents/{filepath}""" + REPO_MILESTONES = """/repos/{owner}/{repo}/milestones""" + + def __init__(self, gitea): + super(Repository, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Repository): return False + return self.owner == other.owner and self.name == other.name + + def __hash__(self): + return hash(self.owner) ^ hash(self.name) + + fields_to_parsers = { + # dont know how to tell apart user and org as owner except form email being empty. + "owner": lambda gitea, r: Organization.parse_response(gitea, r) + if r["email"] == "" + else User.parse_response(gitea, r), + "updated_at": lambda gitea, t: Util.convert_time(t), + } + + @classmethod + def request(cls, gitea, owner, name): + return cls._request(gitea, {"owner": owner, "name": name}) + + patchable_fields = { + "allow_merge_commits", + "allow_rebase", + "allow_rebase_explicit", + "allow_squash_merge", + "archived", + "default_branch", + "description", + "has_issues", + "has_pull_requests", + "has_wiki", + "ignore_whitespace_conflicts", + "name", + "private", + "website", + } + + def get_branches(self) -> List['Branch']: + """Get all the Branches of this Repository.""" + results = self.gitea.requests_get( + Repository.REPO_BRANCHES % (self.owner.username, self.name) + ) + return [Branch.parse_response(self.gitea, result) for result in results] + + def add_branch(self, create_from: Branch, newname: str) -> "Commit": + """Add a branch to the repository""" + # Note: will only work with gitea 1.13 or higher! + data = {"new_branch_name": newname, "old_branch_name": create_from.name} + result = self.gitea.requests_post( + Repository.REPO_BRANCHES % (self.owner.username, self.name), data=data + ) + return Branch.parse_response(self.gitea, result) + + def get_issues(self) -> List["Issue"]: + """Get all Issues of this Repository (open and closed)""" + return self.get_issues_state(Issue.OPENED) + self.get_issues_state(Issue.CLOSED) + + def get_commits(self) -> List["Commit"]: + """Get all the Commits of this Repository.""" + try: + results = self.gitea.requests_get_commits( + Repository.REPO_COMMITS % (self.owner.username, self.name) + ) + except ConflictException as err: + logging.warning(err) + logging.warning( + "Repository %s/%s is Empty" % (self.owner.username, self.name) + ) + results = [] + return [Commit.parse_response(self.gitea, result) for result in results] + + def get_issues_state(self, state) -> List["Issue"]: + """Get issues of state Issue.open or Issue.closed of a repository.""" + assert state in [Issue.OPENED, Issue.CLOSED] + issues = [] + # "page": -1 is returning _all_ issues instead of pages. Hopefully this is intended behaviour. + data = {"page": -1, "state": state} + results = self.gitea.requests_get( + Repository.REPO_ISSUES % (self.owner.username, self.name), params=data + ) + for result in results: + issue = Issue.parse_response(self.gitea, result) + # adding data not contained in the issue answer + setattr(issue, "repo", self.name) + setattr(issue, "owner", self.owner) + issues.append(issue) + return issues + + def get_times(self): + results = self.gitea.requests_get( + Repository.REPO_TIMES % (self.owner.username, self.name) + ) + return results + + def get_user_time(self, username) -> float: + if isinstance(username, User): + username = username.username + results = self.gitea.requests_get( + Repository.REPO_USER_TIME % (self.owner.username, self.name, username) + ) + time = sum(r["time"] for r in results) + return time + + def get_full_name(self) -> str: + return self.owner.username + "/" + self.name + + def create_issue(self, title, assignees=[], description="") -> ApiObject: + data = { + "assignees": assignees, + "body": description, + "closed": False, + "title": title, + } + result = self.gitea.requests_post( + Repository.REPO_ISSUES % (self.owner.username, self.name), data=data + ) + return Issue.parse_response(self.gitea, result) + + def create_milestone(self, title: str, description: str, due_date: str = None, state:str = "open") -> "Milestone": + url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name) + data= {"title": title, "description": description, "state": state} + if due_date: data["due_date"] = due_date + result = self.gitea.requests_post(url, data=data) + return Milestone.parse_response(self.gitea, result) + + def create_gitea_hook(self, hook_url: str, events: List[str]): + url = f"/repos/{self.owner.username}/{self.name}/hooks" + data = { + "type": "gitea", + "config": {"content_type": "json", "url": hook_url}, + "events": events, + "active": True, + } + return self.gitea.requests_post(url, data=data) + + def list_hooks(self): + url = f"/repos/{self.owner.username}/{self.name}/hooks" + return self.gitea.requests_get(url) + + def delete_hook(self, id: str): + url = f"/repos/{self.owner.username}/{self.name}/hooks/{id}" + self.gitea.requests_delete(url) + + def is_collaborator(self, username) -> bool: + if isinstance(username, User): + username = username.username + try: + # returns 204 if its ok, 404 if its not + self.gitea.requests_get( + Repository.REPO_IS_COLLABORATOR + % (self.owner.username, self.name, username) + ) + return True + except: + return False + + def get_users_with_access(self) -> Sequence[User]: + url = f"/repos/{self.owner.username}/{self.name}/collaborators" + response = self.gitea.requests_get(url) + 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 remove_collaborator(self, user_name: str): + url = f"/repos/{self.owner.username}/{self.name}/collaborators/{user_name}" + self.gitea.requests_delete(url) + + def transfer_ownership(self, new_owner: Union["User", "Organization"], new_teams: Set["Team"] = frozenset()): + url = Repository.REPO_TRANSFER.format(owner=self.owner.username, repo=self.name) + data = {"new_owner": new_owner.username} + if isinstance(new_owner, Organization): + new_team_ids = [team.id for team in new_teams if team in new_owner.get_teams()] + data["team_ids"] = new_team_ids + self.gitea.requests_post(url, data=data) + # TODO: make sure this instance is either updated or discarded + + def get_git_content(self: str = None, commit : "Commit" = None) -> List["Content"]: + """https://git.sopranium.de/api/swagger#/repository/repoGetContentsList""" + url = Repository.REPO_CONTENTS.format(owner=self.owner.username, repo=self.name) + data = {"ref": "HEAD" if commit is None else commit.sha} + result = [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)] + return result + + def get_file_content(self, content: "Content", commit : "Commit" = None) -> Union[str, List["Content"]]: + """https://git.sopranium.de/api/swagger#/repository/repoGetContents""" + url = Repository.REPO_CONTENT.format(owner=self.owner.username, + repo=self.name, filepath=content.path) + data = {"ref": "HEAD" if commit is None else commit.sha} + if content.type == Content.FILE: + return self.gitea.requests_get(url, data)["content"] + else: + return [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)] + + def delete(self): + self.gitea.requests_delete( + Repository.REPO_DELETE % (self.owner.username, self.name) + ) + self.deleted = True + +class Milestone(ApiObject): + API_OBJECT = """/repos/{owner}/{repo}/milestones/{number}""" # + + def __init__(self, gitea): + super(Milestone, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Milestone): return False + return self.gitea == other.gitea and self.id == other.id + + def __hash__(self): + return hash(self.gitea) ^ hash(self.id) + + fields_to_parsers = { + "closed_at": lambda gitea, t: Util.convert_time(t), + "due_on": lambda gitea, t: Util.convert_time(t), + } + + patchable_fields = { + "allow_merge_commits", + "allow_rebase", + "allow_rebase_explicit", + "allow_squash_merge", + "archived", + "default_branch", + "description", + "has_issues", + "has_pull_requests", + "has_wiki", + "ignore_whitespace_conflicts", + "name", + "private", + "website", + } + + @classmethod + def request(cls, gitea, owner, repo, number): + return cls._request(gitea, {"owner": owner, "repo": repo, "number": number}) + + +class Comment(ApiObject): + + def __init__(self, gitea): + super(Comment, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Comment): return False + return self.repo == other.repo and self.id == other.id + + def __hash__(self): + return hash(self.repo) ^ hash(self.id) + + fields_to_parsers = { + "user": lambda gitea, r: User.parse_response(gitea, r), + "created_at": lambda gitea, t: Util.convert_time(t), + "updated_at": lambda gitea, t: Util.convert_time(t), + } + + +class Commit(ReadonlyApiObject): + def __init__(self, gitea): + super(Commit, self).__init__(gitea) + + fields_to_parsers = { + # NOTE: api may return None for commiters that are no gitea users + "author": lambda gitea, u: User.parse_response(gitea, u) if u else None + } + + def __eq__(self, other): + if not isinstance(other, Commit): return False + return self.sha == other.sha + + def __hash__(self): + return hash(self.sha) + + @classmethod + def parse_response(cls, gitea, result): + commit_cache = result["commit"] + api_object = cls(gitea) + cls._initialize(gitea, api_object, result) + # inner_commit for legacy reasons + Commit._add_read_property("inner_commit", commit_cache, api_object) + return api_object + + +class Issue(ApiObject): + API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # + GET_TIME = """/repos/%s/%s/issues/%s/times""" # + GET_COMMENTS = """/repos/%s/%s/issues/comments""" + CREATE_ISSUE = """/repos/{owner}/{repo}/issues""" + + OPENED = "open" + CLOSED = "closed" + + def __init__(self, gitea): + super(Issue, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Issue): return False + return self.repo == other.repo and self.id == other.id + + def __hash__(self): + return hash(self.repo) ^ hash(self.id) + + fields_to_parsers = { + "milestone": lambda gitea, m: Milestone.parse_response(gitea, m), + "user": lambda gitea, u: User.parse_response(gitea, u), + "assignee": lambda gitea, u: User.parse_response(gitea, u), + "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, + } + + patchable_fields = { + "assignee", + "assignees", + "body", + "due_date", + "milestone", + "state", + "title", + } + + @classmethod + def request(cls, gitea, owner, repo, number): + api_object = cls._request(gitea, {"owner": owner, "repo": repo, "number": number}) + return api_object + + @classmethod + def create_issue(cls, gitea, repo: Repository, title: str, body: str = ""): + args = {"owner": repo.owner.username, "repo": repo.name} + data = {"title": title, "body": body} + result = gitea.requests_post(Issue.CREATE_ISSUE.format(**args), data=data) + return Issue.parse_response(gitea, result) + + def get_time_sum(self, user: User) -> int: + results = self.gitea.requests_get( + Issue.GET_TIME % (self.owner.username, self.repo, self.number) + ) + return sum( + result["time"] + for result in results + if result and result["user_id"] == user.id + ) + + def get_times(self) -> Optional[Dict]: + return self.gitea.requests_get( + Issue.GET_TIME % (self.owner.username, self.repo, self.number) + ) + + def delete_time(self, time_id: str): + path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times/{time_id}" + self.gitea.requests_delete(path) + + def add_time(self, time: int, created: str = None, user_name: str = None): + path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times" + self.gitea.requests_post( + path, data={"created": created, "time": int(time), "user_name": user_name} + ) + + def get_comments(self) -> List[ApiObject]: + results = self.gitea.requests_get( + Issue.GET_COMMENTS % (self.owner.username, self.repo) + ) + 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 + return [ + comment + for comment in allProjectComments + if comment.issue_url.endswith("/" + str(self.number)) + ] + + +class Team(ApiObject): + API_OBJECT = """/teams/{id}""" # + ADD_USER = """/teams/%s/members/%s""" # + ADD_REPO = """/teams/%s/repos/%s/%s""" # + TEAM_DELETE = """/teams/%s""" # + GET_MEMBERS = """/teams/%s/members""" # + GET_REPOS = """/teams/%s/repos""" # + + def __init__(self, gitea): + super(Team, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Team): return False + return self.organization == other.organization and self.id == other.id + + def __hash__(self): + return hash(self.organization) ^ hash(self.id) + + fields_to_parsers = { + "organization": lambda gitea, o: Organization.parse_response(gitea, o) + } + + @classmethod + def request(cls, gitea, organization, team): + return cls._request(gitea, {"id": id}) + + patchable_fields = {"description", "name", "permission", "units"} + + def add_user(self, user: User): + self.gitea.requests_put(Team.ADD_USER % (self.id, user.login)) + + def add_repo(self, org: Organization, repo: Repository): + self.gitea.requests_put(Team.ADD_REPO % (self.id, org, repo.name)) + + def get_members(self): + """ Get all users assigned to the team. """ + results = self.gitea.requests_get(Team.GET_MEMBERS % self.id) + return [User.parse_response(self.gitea, result) for result in results] + + def get_repos(self): + """ Get all repos of this Team.""" + results = self.gitea.requests_get(Team.GET_REPOS % self.id) + return [Repository.parse_response(self.gitea, result) for result in results] + + def delete(self): + self.gitea.requests_delete(Team.TEAM_DELETE % self.id) + self.deleted = True + + def remove_team_member(self, user_name: str): + url = f"/teams/{self.id}/members/{user_name}" + self.gitea.requests_delete(url) + +class Content(ReadonlyApiObject): + FILE = "file" + + def __init__(self, gitea): + super(Content, self).__init__(gitea) + + def __eq__(self, other): + if not isinstance(other, Team): return False + return self.repo == self.repo and self.sha == other.sha and self.name == other.name + + def __hash__(self): + return hash(self.repo) ^ hash(self.sha) ^ hash(self.name) + + +class Util: + @staticmethod + def convert_time(time: str) -> datetime: + """ Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)""" + try: + return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z") + except ValueError: + return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S") + diff --git a/gitea/baseapiobject.py b/gitea/baseapiobject.py index 89fca9a..a9499a8 100644 --- a/gitea/baseapiobject.py +++ b/gitea/baseapiobject.py @@ -1,7 +1,5 @@ from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation,RawRequestEndpointMissing - - class ReadonlyApiObject: def __init__(self, gitea): @@ -23,7 +21,7 @@ class ReadonlyApiObject: @classmethod def request(cls, gitea, id): - if hasattr("GET_API_OBJECT", cls): + if hasattr("API_OBJECT", cls): return cls._request(gitea) else: raise RawRequestEndpointMissing() @@ -41,7 +39,7 @@ class ReadonlyApiObject: @classmethod def _get_gitea_api_object(cls, gitea, args): """Retrieving an object always as GET_API_OBJECT """ - return gitea.requests_get(cls.GET_API_OBJECT.format(**args)) + return gitea.requests_get(cls.API_OBJECT.format(**args)) @classmethod def parse_response(cls, gitea, result) -> "ReadonlyApiObject": diff --git a/gitea/gitea.py b/gitea/gitea.py index 037162c..f7f1906 100644 --- a/gitea/gitea.py +++ b/gitea/gitea.py @@ -1,746 +1,15 @@ import json import logging -from datetime import datetime -from typing import List, Tuple, Dict, Sequence, Optional, Union, Set +from typing import List, Dict import requests from httpcache import CachingHTTPAdapter -from .baseapiobject import ReadonlyApiObject, ApiObject -from .exceptions import * - - -class Organization(ApiObject): - """see https://try.gitea.io/api/swagger#/organization/orgGetAll""" - - GET_API_OBJECT = """/orgs/{name}""" # - PATCH_API_OBJECT = """/orgs/{name}""" # - ORG_REPOS_REQUEST = """/orgs/%s/repos""" # - ORG_TEAMS_REQUEST = """/orgs/%s/teams""" # - ORG_TEAMS_CREATE = """/orgs/%s/teams""" # - ORG_GET_MEMBERS = """/orgs/%s/members""" # - ORG_IS_MEMBER = """/orgs/%s/members/%s""" # , - ORG_DELETE = """/orgs/%s""" # - ORG_HEATMAP = """/users/%s/heatmap""" # - - def __init__(self, gitea): - super(Organization, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Organization): return False - return self.gitea == other.gitea and self.name == other.name - - def __hash__(self): - return hash(self.gitea) ^ hash(self.name) - - @classmethod - def request(cls, gitea, name): - return cls._request(gitea, {"name": name}) - - @classmethod - def parse_response(cls, gitea, result): - api_object = super().parse_response(gitea, result) - # add "name" field to make this behave similar to users - Organization._add_read_property("name", result["username"], api_object) - return api_object - - patchable_fields = {"description", "full_name", "location", "visibility", "website"} - - def commit(self): - values = self.get_dirty_fields() - args = {"name": self.name} - self.gitea.requests_patch( - Organization.PATCH_API_OBJECT.format(**args), data=values - ) - self.dirty_fields = {} - - def get_repositories(self) -> List["Repository"]: - results = self.gitea.requests_get_paginated( - Organization.ORG_REPOS_REQUEST % self.username - ) - return [Repository.parse_response(self.gitea, result) for result in results] - - def get_repository(self, name) -> "Repository": - repos = self.get_repositories() - for repo in repos: - if repo.name == name: - return repo - raise NotFoundException("Repository %s not existent in organization." % name) - - def get_teams(self) -> List["Team"]: - results = self.gitea.requests_get( - Organization.ORG_TEAMS_REQUEST % self.username - ) - teams = [Team.parse_response(self.gitea, result) for result in results] - # organisation seems to be missing using this request, so we add org manually - for t in teams: setattr(t, "_organization", self) - return teams - - def get_team(self, name) -> "Team": - teams = self.get_teams() - for team in teams: - if team.name == name: - return team - raise NotFoundException("Team not existent in organization.") - - def get_members(self) -> List["User"]: - results = self.gitea.requests_get(Organization.ORG_GET_MEMBERS % self.username) - return [User.parse_response(self.gitea, result) for result in results] - - def is_member(self, username) -> bool: - if isinstance(username, User): - username = username.username - try: - # returns 204 if its ok, 404 if its not - self.gitea.requests_get( - Organization.ORG_IS_MEMBER % (self.username, username) - ) - return True - except: - return False - - def remove_member(self, user: "User"): - path = f"/orgs/{self.username}/members/{user.username}" - self.gitea.requests_delete(path) - - def delete(self): - """ Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User""" - for repo in self.get_repositories(): - repo.delete() - self.gitea.requests_delete(Organization.ORG_DELETE % self.username) - self.deleted = True - - def get_heatmap(self) -> List[Tuple[datetime, int]]: - results = self.gitea.requests_get(User.USER_HEATMAP % self.username) - results = [ - (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) - for result in results - ] - return results - - -class User(ApiObject): - GET_API_OBJECT = """/users/{name}""" # - USER_MAIL = """/user/emails?sudo=%s""" # - USER_PATCH = """/admin/users/%s""" # - ADMIN_DELETE_USER = """/admin/users/%s""" # - ADMIN_EDIT_USER = """/admin/users/{username}""" # - USER_HEATMAP = """/users/%s/heatmap""" # - - def __init__(self, gitea): - super(User, self).__init__(gitea) - self._emails = [] - - def __eq__(self, other): - if not isinstance(other, User): return False - return self.gitea == other.gitea and self.id == other.id - - def __hash__(self): - return hash(self.gitea) ^ hash(self.id) - - @property - def emails(self): - self.__request_emails() - return self._emails - - @classmethod - def request(cls, gitea, name) -> "User": - api_object = cls._request(gitea, {"name": name}) - return api_object - - patchable_fields = { - "active", - "admin", - "allow_create_organization", - "allow_git_hook", - "allow_import_local", - "email", - "full_name", - "location", - "login_name", - "max_repo_creation", - "must_change_password", - "password", - "prohibit_login", - "website", - } - - def commit(self, login_name: str, source_id: int = 0): - """ - Unfortunately it is necessary to require the login name - as well as the login source (that is not supplied when getting a user) for - changing a user. - Usually source_id is 0 and the login_name is equal to the username. - """ - values = self.get_dirty_fields() - values.update( - # api-doc says that the "source_id" is necessary; works without though - {"login_name": login_name, "source_id": source_id} - ) - args = {"username": self.username} - self.gitea.requests_patch(User.ADMIN_EDIT_USER.format(**args), data=values) - self.dirty_fields = {} - - def get_repositories(self) -> List["Repository"]: - """ Get all Repositories owned by this User.""" - url = f"/users/{self.username}/repos" - results = self.gitea.requests_get(url) - return [Repository.parse_response(self.gitea, result) for result in results] - - def get_orgs(self) -> List[Organization]: - """ Get all Organizations this user is a member of.""" - url = f"/users/{self.username}/orgs" - results = self.gitea.requests_get(url) - return [Organization.parse_response(self.gitea, result) for result in results] - - def get_teams(self) -> List['Team']: - url = f"/user/teams" - results = self.gitea.requests_get(url, sudo=self) - return [Team.parse_response(self.gitea, result) for result in results] - - def get_accessible_repos(self) -> List['Repository']: - """ Get all Repositories accessible by the logged in User.""" - results = self.gitea.requests_get("/user/repos", sudo=self) - return [Repository.parse_response(self, result) for result in results] - - def __request_emails(self): - result = self.gitea.requests_get(User.USER_MAIL % self.login) - # report if the adress changed by this - for mail in result: - self._emails.append(mail["email"]) - if mail["primary"]: - self._email = mail["email"] - - def delete(self): - """ Deletes this User. Also deletes all Repositories he owns.""" - self.gitea.requests_delete(User.ADMIN_DELETE_USER % self.username) - self.deleted = True - - def get_heatmap(self) -> List[Tuple[datetime, int]]: - results = self.gitea.requests_get(User.USER_HEATMAP % self.username) - results = [ - (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) - for result in results - ] - return results - - -class Branch(ReadonlyApiObject): - GET_API_OBJECT = """/repos/%s/%s/branches/%s""" # , , - - def __init__(self, gitea): - super(Branch, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Branch): - return False - return self.commit == other.commit and self.name == other.name - - def __hash__(self): - return hash(self.commit["id"]) ^ hash(self.name) - - fields_to_parsers = { - # This is not a commit object - #"commit": lambda gitea, c: Commit.parse_response(gitea, c) - } - - @classmethod - def request(cls, gitea, owner, repo, ref): - return cls._request(gitea, {"owner": owner, "repo": repo, "ref": ref}) - - -class Repository(ApiObject): - REPO_IS_COLLABORATOR = ( - """/repos/%s/%s/collaborators/%s""" # , , - ) - GET_API_OBJECT = """/repos/{owner}/{name}""" # , - REPO_SEARCH = """/repos/search/%s""" # - REPO_BRANCHES = """/repos/%s/%s/branches""" # , - REPO_ISSUES = """/repos/%s/%s/issues""" # - REPO_DELETE = """/repos/%s/%s""" # , - REPO_TIMES = """/repos/%s/%s/times""" # , - REPO_USER_TIME = """/repos/%s/%s/times/%s""" # , , - REPO_COMMITS = "/repos/%s/%s/commits" # , - REPO_TRANSFER = "/repos/{owner}/{repo}/transfer" - REPO_CONTENTS = "/repos/{owner}/{repo}/contents" - REPO_CONTENT = """/repos/{owner}/{repo}/contents/{filepath}""" - REPO_MILESTONES = """/repos/{owner}/{repo}/milestones""" - - def __init__(self, gitea): - super(Repository, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Repository): return False - return self.owner == other.owner and self.name == other.name - - def __hash__(self): - return hash(self.owner) ^ hash(self.name) - - fields_to_parsers = { - # dont know how to tell apart user and org as owner except form email being empty. - "owner": lambda gitea, r: Organization.parse_response(gitea, r) - if r["email"] == "" - else User.parse_response(gitea, r), - "updated_at": lambda gitea, t: Util.convert_time(t), - } - - @classmethod - def request(cls, gitea, owner, name): - return cls._request(gitea, {"owner": owner, "name": name}) - - patchable_fields = { - "allow_merge_commits", - "allow_rebase", - "allow_rebase_explicit", - "allow_squash_merge", - "archived", - "default_branch", - "description", - "has_issues", - "has_pull_requests", - "has_wiki", - "ignore_whitespace_conflicts", - "name", - "private", - "website", - } - - def get_branches(self) -> List['Branch']: - """Get all the Branches of this Repository.""" - results = self.gitea.requests_get( - Repository.REPO_BRANCHES % (self.owner.username, self.name) - ) - return [Branch.parse_response(self.gitea, result) for result in results] - - def add_branch(self, create_from: Branch, newname: str) -> "Commit": - """Add a branch to the repository""" - # Note: will only work with gitea 1.13 or higher! - data = {"new_branch_name": newname, "old_branch_name": create_from.name} - result = self.gitea.requests_post( - Repository.REPO_BRANCHES % (self.owner.username, self.name), data=data - ) - return Branch.parse_response(self.gitea, result) - - def get_issues(self) -> List["Issue"]: - """Get all Issues of this Repository (open and closed)""" - return self.get_issues_state(Issue.OPENED) + self.get_issues_state(Issue.CLOSED) - - def get_commits(self) -> List["Commit"]: - """Get all the Commits of this Repository.""" - try: - results = self.gitea.requests_get_commits( - Repository.REPO_COMMITS % (self.owner.username, self.name) - ) - except ConflictException as err: - logging.warning(err) - logging.warning( - "Repository %s/%s is Empty" % (self.owner.username, self.name) - ) - results = [] - return [Commit.parse_response(self.gitea, result) for result in results] - - def get_issues_state(self, state) -> List["Issue"]: - """Get issues of state Issue.open or Issue.closed of a repository.""" - assert state in [Issue.OPENED, Issue.CLOSED] - issues = [] - # "page": -1 is returning _all_ issues instead of pages. Hopefully this is intended behaviour. - data = {"page": -1, "state": state} - results = self.gitea.requests_get( - Repository.REPO_ISSUES % (self.owner.username, self.name), params=data - ) - for result in results: - issue = Issue.parse_response(self.gitea, result) - # adding data not contained in the issue answer - setattr(issue, "repo", self.name) - setattr(issue, "owner", self.owner) - issues.append(issue) - return issues - - def get_times(self): - results = self.gitea.requests_get( - Repository.REPO_TIMES % (self.owner.username, self.name) - ) - return results - - def get_user_time(self, username) -> float: - if isinstance(username, User): - username = username.username - results = self.gitea.requests_get( - Repository.REPO_USER_TIME % (self.owner.username, self.name, username) - ) - time = sum(r["time"] for r in results) - return time - - def get_full_name(self) -> str: - return self.owner.username + "/" + self.name - - def create_issue(self, title, assignees=[], description="") -> ApiObject: - data = { - "assignees": assignees, - "body": description, - "closed": False, - "title": title, - } - result = self.gitea.requests_post( - Repository.REPO_ISSUES % (self.owner.username, self.name), data=data - ) - return Issue.parse_response(self.gitea, result) - - def create_milestone(self, title: str, description: str, due_date: str = None, state:str = "open") -> "Milestone": - url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name) - data= {"title": title, "description": description, "state": state} - if due_date: data["due_date"] = due_date - result = self.gitea.requests_post(url, data=data) - return Milestone.parse_response(self.gitea, result) - - def create_gitea_hook(self, hook_url: str, events: List[str]): - url = f"/repos/{self.owner.username}/{self.name}/hooks" - data = { - "type": "gitea", - "config": {"content_type": "json", "url": hook_url}, - "events": events, - "active": True, - } - return self.gitea.requests_post(url, data=data) - - def list_hooks(self): - url = f"/repos/{self.owner.username}/{self.name}/hooks" - return self.gitea.requests_get(url) - - def delete_hook(self, id: str): - url = f"/repos/{self.owner.username}/{self.name}/hooks/{id}" - self.gitea.requests_delete(url) - - def is_collaborator(self, username) -> bool: - if isinstance(username, User): - username = username.username - try: - # returns 204 if its ok, 404 if its not - self.gitea.requests_get( - Repository.REPO_IS_COLLABORATOR - % (self.owner.username, self.name, username) - ) - return True - except: - return False - - def get_users_with_access(self) -> Sequence[User]: - url = f"/repos/{self.owner.username}/{self.name}/collaborators" - response = self.gitea.requests_get(url) - 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 remove_collaborator(self, user_name: str): - url = f"/repos/{self.owner.username}/{self.name}/collaborators/{user_name}" - self.gitea.requests_delete(url) - - def transfer_ownership(self, new_owner: Union["User", "Organization"], new_teams: Set["Team"] = frozenset()): - url = Repository.REPO_TRANSFER.format(owner=self.owner.username, repo=self.name) - data = {"new_owner": new_owner.username} - if isinstance(new_owner, Organization): - new_team_ids = [team.id for team in new_teams if team in new_owner.get_teams()] - data["team_ids"] = new_team_ids - self.gitea.requests_post(url, data=data) - # TODO: make sure this instance is either updated or discarded - - def get_git_content(self: str = None, commit : "Commit" = None) -> List["Content"]: - """https://git.sopranium.de/api/swagger#/repository/repoGetContentsList""" - url = Repository.REPO_CONTENTS.format(owner=self.owner.username, repo=self.name) - data = {"ref": "HEAD" if commit is None else commit.sha} - result = [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)] - return result - - def get_file_content(self, content: "Content", commit : "Commit" = None) -> Union[str, List["Content"]]: - """https://git.sopranium.de/api/swagger#/repository/repoGetContents""" - url = Repository.REPO_CONTENT.format(owner=self.owner.username, - repo=self.name, filepath=content.path) - data = {"ref": "HEAD" if commit is None else commit.sha} - if content.type == Content.FILE: - return self.gitea.requests_get(url, data)["content"] - else: - return [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)] - - def delete(self): - self.gitea.requests_delete( - Repository.REPO_DELETE % (self.owner.username, self.name) - ) - self.deleted = True - -class Milestone(ApiObject): - GET_API_OBJECT = ( - """/repos/{owner}/{repo}/milestones/{number}""" # - ) - - def __init__(self, gitea): - super(Milestone, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Milestone): return False - return self.gitea == other.gitea and self.id == other.id - - def __hash__(self): - return hash(self.gitea) ^ hash(self.id) - - fields_to_parsers = { - "closed_at": lambda gitea, t: Util.convert_time(t), - "due_on": lambda gitea, t: Util.convert_time(t), - } - - patchable_fields = { - "allow_merge_commits", - "allow_rebase", - "allow_rebase_explicit", - "allow_squash_merge", - "archived", - "default_branch", - "description", - "has_issues", - "has_pull_requests", - "has_wiki", - "ignore_whitespace_conflicts", - "name", - "private", - "website", - } - - @classmethod - def request(cls, gitea, owner, repo, number): - return cls._request(gitea, {"owner": owner, "repo": repo, "number": number}) - - -class Comment(ApiObject): - PATCH_API_OBJECT = "/repos/{owner}/{repo}/issues/comments/{id}" - - def __init__(self, gitea): - super(Comment, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Comment): return False - return self.repo == other.repo and self.id == other.id - - def __hash__(self): - return hash(self.repo) ^ hash(self.id) - - fields_to_parsers = { - "user": lambda gitea, r: User.parse_response(gitea, r), - "created_at": lambda gitea, t: Util.convert_time(t), - "updated_at": lambda gitea, t: Util.convert_time(t), - } - - patchable_fields = {"body"} - - -class Commit(ReadonlyApiObject): - def __init__(self, gitea): - super(Commit, self).__init__(gitea) - - fields_to_parsers = { - # NOTE: api may return None for commiters that are no gitea users - "author": lambda gitea, u: User.parse_response(gitea, u) if u else None - } - - def __eq__(self, other): - if not isinstance(other, Commit): return False - return self.sha == other.sha - - def __hash__(self): - return hash(self.sha) - - @classmethod - def request(cls, gitea, owner, repo): - api_object = cls._request(gitea, {"owner": owner, "repo": repo}) - return api_object - - @classmethod - def parse_response(cls, gitea, result): - commit_cache = result["commit"] - api_object = cls(gitea) - cls._initialize(gitea, api_object, result) - Commit._add_read_property("inner_commit", commit_cache, api_object) - return api_object - - -class Issue(ApiObject): - GET_API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # - GET_TIME = """/repos/%s/%s/issues/%s/times""" # - GET_COMMENTS = """/repos/%s/%s/issues/comments""" - CREATE_ISSUE = """/repos/{owner}/{repo}/issues""" - - OPENED = "open" - CLOSED = "closed" - - def __init__(self, gitea): - super(Issue, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Issue): return False - return self.repo == other.repo and self.id == other.id - - def __hash__(self): - return hash(self.repo) ^ hash(self.id) - - fields_to_parsers = { - "milestone": lambda gitea, m: Milestone.parse_response(gitea, m), - "user": lambda gitea, u: User.parse_response(gitea, u), - "assignee": lambda gitea, u: User.parse_response(gitea, u), - "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, - } - - patchable_fields = { - "assignee", - "assignees", - "body", - "due_date", - "milestone", - "state", - "title", - } - - @classmethod - def request(cls, gitea, owner, repo, number): - api_object = cls._request( - gitea, {"owner": owner, "repo": repo, "number": number} - ) - return api_object - - @classmethod - def create_issue(cls, gitea, repo: Repository, title: str, body: str = ""): - args = {"owner": repo.owner.username, "repo": repo.name} - data = {"title": title, "body": body} - result = gitea.requests_post(Issue.CREATE_ISSUE.format(**args), data=data) - return Issue.parse_response(gitea, result) - - def get_time_sum(self, user: User) -> int: - results = self.gitea.requests_get( - Issue.GET_TIME % (self.owner.username, self.repo, self.number) - ) - return sum( - result["time"] - for result in results - if result and result["user_id"] == user.id - ) - - def get_times(self) -> Optional[Dict]: - return self.gitea.requests_get( - Issue.GET_TIME % (self.owner.username, self.repo, self.number) - ) - - def delete_time(self, time_id: str): - path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times/{time_id}" - self.gitea.requests_delete(path) - - def add_time(self, time: int, created: str = None, user_name: str = None): - path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times" - self.gitea.requests_post( - path, data={"created": created, "time": int(time), "user_name": user_name} - ) - - def get_comments(self) -> List[ApiObject]: - results = self.gitea.requests_get( - Issue.GET_COMMENTS % (self.owner.username, self.repo) - ) - 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 - return [ - comment - for comment in allProjectComments - if comment.issue_url.endswith("/" + str(self.number)) - ] - - -class Team(ApiObject): - GET_API_OBJECT = """/teams/{id}""" # - ADD_USER = """/teams/%s/members/%s""" # - ADD_REPO = """/teams/%s/repos/%s/%s""" # - TEAM_DELETE = """/teams/%s""" # - GET_MEMBERS = """/teams/%s/members""" # - GET_REPOS = """/teams/%s/repos""" # - - def __init__(self, gitea): - super(Team, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Team): return False - return self.organization == other.organization and self.id == other.id - - def __hash__(self): - return hash(self.organization) ^ hash(self.id) - - fields_to_parsers = { - "organization": lambda gitea, o: Organization.parse_response(gitea, o) - } - - @classmethod - def request(cls, gitea, organization, team): - return cls._request(gitea, {"id": id}) - - patchable_fields = {"description", "name", "permission", "units"} - - def add_user(self, user: User): - self.gitea.requests_put(Team.ADD_USER % (self.id, user.login)) - - def add_repo(self, org: Organization, repo: Repository): - self.gitea.requests_put(Team.ADD_REPO % (self.id, org, repo.name)) - - def get_members(self): - """ Get all users assigned to the team. """ - results = self.gitea.requests_get(Team.GET_MEMBERS % self.id) - return [User.parse_response(self.gitea, result) for result in results] - - def get_repos(self): - """ Get all repos of this Team.""" - results = self.gitea.requests_get(Team.GET_REPOS % self.id) - return [Repository.parse_response(self.gitea, result) for result in results] - - def delete(self): - self.gitea.requests_delete(Team.TEAM_DELETE % self.id) - self.deleted = True - - def remove_team_member(self, user_name: str): - url = f"/teams/{self.id}/members/{user_name}" - self.gitea.requests_delete(url) - -class Content(ReadonlyApiObject): - GET_API_OBJECT = """/repos/{owner}/{repo}/contents/{filepath}""" - - FILE = "file" - - def __init__(self, gitea): - super(Content, self).__init__(gitea) - - def __eq__(self, other): - if not isinstance(other, Team): return False - return self.repo == self.repo and self.sha == other.sha and self.name == other.name - - def __hash__(self): - return hash(self.repo) ^ hash(self.sha) ^ hash(self.name) - - -class Util: - @staticmethod - def convert_time(time: str) -> datetime: - """ Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)""" - try: - return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z") - except ValueError: - return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S") - +from .exceptions import NotFoundException, ConflictException, AlreadyExistsException +from .apiobject import User, Organization, Repository, Team class Gitea: """ Object to establish a session with Gitea. """ - ADMIN_CREATE_USER = """/admin/users""" GET_USERS_ADMIN = """/admin/users""" ADMIN_REPO_CREATE = """/admin/users/%s/repos""" # diff --git a/setup.py b/setup.py index 2d2aedb..7978a90 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ with open('README.md') as readme_file: setup_args = dict( name='py-gitea', - version='0.1.8', + version='0.1.9', description='A python wrapper for the Gitea API', long_description_content_type="text/markdown", long_description=README, diff --git a/tests/test_api.py b/tests/test_api.py index 4030386..184d6c6 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -282,7 +282,10 @@ def test_delete_org(instance): def test_delete_user(instance): - user = User.request(instance, test_user) + user_name = test_user + "delte_test" + email = user_name + "@example.org" + user = instance.create_user(user_name, email, "abcdefg1.23AB", send_notify=False) + assert user.username == user_name user.delete() with pytest.raises(NotFoundException) as e: - User.request(instance, test_user) + User.request(instance, user_name) From d1430acf8a441d4430e7f8d5537c4e603b9a4eea Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Mon, 15 Nov 2021 17:39:11 +0100 Subject: [PATCH 03/13] removed legacy stuff (hopefully this will not break anything) --- gitea/apiobject.py | 2 +- gitea/gitea.py | 278 +++++++-------------------------------------- 2 files changed, 39 insertions(+), 241 deletions(-) diff --git a/gitea/apiobject.py b/gitea/apiobject.py index 38fc24f..88e53ae 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -318,7 +318,7 @@ class Repository(ApiObject): def get_commits(self) -> List["Commit"]: """Get all the Commits of this Repository.""" try: - results = self.gitea.requests_get_commits( + results = self.gitea.requests_get_paginated( Repository.REPO_COMMITS % (self.owner.username, self.name) ) except ConflictException as err: diff --git a/gitea/gitea.py b/gitea/gitea.py index f7f1906..7074c39 100644 --- a/gitea/gitea.py +++ b/gitea/gitea.py @@ -1,8 +1,9 @@ import json import logging -from typing import List, Dict +from typing import List, Dict, Union import requests +from frozendict import frozendict from httpcache import CachingHTTPAdapter from .exceptions import NotFoundException, ConflictException, AlreadyExistsException @@ -44,158 +45,72 @@ class Gitea: return json.loads(result.text) return {} - def requests_get_paginated(self, endpoint, params ={}, requests=None, sudo=None, page_key: str = "page"): + def requests_get(self, endpoint:str , params= frozendict(), sudo=None): + combined_params = {} + combined_params.update(params) + if sudo: + combined_params["sudo"] = sudo.username + request = self.requests.get(self.__get_url(endpoint), headers=self.headers, params=combined_params) + if request.status_code == 204: + return None + if request.status_code not in [200, 201]: + message = f"Received status code: {request.status_code} ({request.url})" + if request.status_code in [404]: + raise NotFoundException(message) + if request.status_code in [403]: + raise Exception(f"Unauthorized: {request.url} - Check your permissions and try again! ({message})") + if request.status_code in [409]: + raise ConflictException(message) + raise Exception(message) + return self.parse_result(request) + + def requests_get_paginated(self, endpoint:str , params = frozendict(), sudo=None, page_key: str = "page"): page = 1 combined_params = {} combined_params.update(params) aggregated_result = [] while True: combined_params[page_key] = page - result = self.requests_get(endpoint, combined_params, requests, sudo) + result = self.requests_get(endpoint, combined_params, sudo) aggregated_result.extend(result) page += 1 if len(result) == 0: return aggregated_result - def requests_get(self, endpoint, params={}, requests=None, sudo=None): - combined_params = {} - combined_params.update(params) - if sudo: - combined_params["sudo"] = sudo.username - if not requests: - request = self.requests.get( - self.__get_url(endpoint), headers=self.headers, params=combined_params - ) - else: - request = requests.get( - self.__get_url(endpoint), headers=self.headers, params=combined_params - ) - if request.status_code == 204: - return None - if request.status_code not in [200, 201]: - message = "Received status code: %s (%s)" % ( - request.status_code, - request.url, - ) - if request.status_code in [404]: - raise NotFoundException(message) - if request.status_code in [403]: - raise Exception( - "Unauthorized: %s - Check your permissions and try again! (%s)" - % (request.url, message) - ) - if request.status_code in [409]: - raise ConflictException(message) - raise Exception(message) - return self.parse_result(request) - - def requests_get_commits(self, endpoint, page=1, params={}, requests=None): - results = [] - page_endpoint = endpoint + f"?page={page}" - if not requests: - request = self.requests.get( - self.__get_url(page_endpoint), headers=self.headers, params=params - ) - else: - request = requests.get( - self.__get_url(page_endpoint), headers=self.headers, params=params - ) - results += self.parse_result(request) - if request.headers.get("x-hasmore") == "true": - page += 1 - results += self.requests_get_commits(endpoint, page) - elif request.status_code not in [200, 201]: - message = "Received status code: %s (%s)" % ( - request.status_code, - request.url, - ) - if request.status_code in [404]: - raise NotFoundException(message) - if request.status_code in [403]: - raise Exception( - "Unauthorized: %s - Check your permissions and try again! (%s)" - % (request.url, message) - ) - if request.status_code in [409]: - raise ConflictException(message) - raise Exception(message) - return results - - def requests_put(self, endpoint): + def requests_put(self, endpoint: str): request = self.requests.put(self.__get_url(endpoint), headers=self.headers) if request.status_code not in [204]: - message = "Received status code: %s (%s) %s" % ( - request.status_code, - request.url, - request.text, - ) + message = f"Received status code: {request.status_code} ({request.url}) {request.text}" self.logger.error(message) raise Exception(message) - def requests_delete(self, endpoint): + def requests_delete(self, endpoint: str): request = self.requests.delete(self.__get_url(endpoint), headers=self.headers) if request.status_code not in [204]: - message = "Received status code: %s (%s)" % ( - request.status_code, - request.url, - ) + message = f"Received status code: {request.status_code} ({request.url})" self.logger.error(message) raise Exception(message) - def requests_post(self, endpoint, data): - """ Post data to API-endpoint. - - Args: - endpoint (str): endpoint of API to send data to - data (dict): Data to send. - - Returns: (dict) - Parsed JSON-answer from the API. - - Throws: - AlreadyExistsException, if 'already exists' in answer - Exception, if status code not ok - """ - request = self.requests.post( - self.__get_url(endpoint), headers=self.headers, data=json.dumps(data) - ) + def requests_post(self, endpoint:str , data:dict): + request = self.requests.post(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)) if request.status_code not in [200, 201, 202]: - 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): self.logger.warning(request.text) raise AlreadyExistsException() - self.logger.error( - "Received status code: %s (%s)" % (request.status_code, request.url) - ) - self.logger.error("With info: %s (%s)" % (data, self.headers)) - self.logger.error("Answer: %s" % request.text) - raise Exception( - "Received status code: %s (%s), %s" - % (request.status_code, request.url, request.text) - ) - + self.logger.error(f"Received status code: {request.status_code} ({ request.url})") + self.logger.error(f"With info: {data} ({self.headers})") + self.logger.error(f"Answer: {request.text}") + raise Exception(f"Received status code: {request.status_code} ({request.url}), {request.text}") return self.parse_result(request) - def requests_patch(self, endpoint, data): - request = self.requests.patch( - self.__get_url(endpoint), headers=self.headers, data=json.dumps(data) - ) + def requests_patch(self, endpoint:str, data:dict): + request = self.requests.patch(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)) if request.status_code not in [200, 201]: - error_message = "Received status code: %s (%s) %s" % ( - request.status_code, - request.url, - data, - ) + error_message = f"Received status code: {request.status_code} ({request.url}) {data}" self.logger.error(error_message) raise Exception(error_message) return self.parse_result(request) - def get_users_search(self): - path = "/users/search" - return self.requests_get(path) - def get_orgs_public_members_all(self, orgname): path = "/orgs/" + orgname + "/public_members" return self.requests_get(path) @@ -205,123 +120,6 @@ class Gitea: results = self.requests_get(path) return [Organization.parse_response(self, result) for result in results] - def post_repos__forks(self, organization, repo, owner): - path = "/repos/" + owner + "/" + repo + "/forks" - return self.requests_post(path, data={"organization": organization}) - - def get_repos_forks(self, repo, owner): - path = "/repos/" + owner + "/" + repo + "/forks" - return self.requests_get(path) - - def put_repos__subscription(self, username, reponame): - path = "/repos/" + username + "/" + reponame + "/subscription" - return self.requests.put(path) - - def delete_repos_subscription(self, username, reponame): - path = "/repos/" + username + "/" + reponame + "/subscription" - return self.requests.delete(path) - - def get_repos_subscription(self, username, reponame): - path = "/repos/" + username + "/" + reponame + "/subscription" - return self.requests_get(path) - - def get_users_following(self, username): - path = "/users/" + username + "/following" - return self.requests_get(path) - - def get_users_starred(self, username): - path = "/users/" + username + "/starred" - return self.requests_get(path) - - def put_orgs_public_members(self, username, orgname): - path = "/orgs/" + orgname + "/public_members/" + username - return self.requests.put(path) - - def delete_orgs_public_members(self, username, orgname): - path = "/orgs/" + orgname + "/public_members/" + username - return self.requests.delete(path) - - def get_orgs_public_members(self, username, orgname): - path = "/orgs/" + orgname + "/public_members/" + username - return self.requests_get(path) - - def post_org_repos( - self, name, description, private, auto_init, gitignores, license, readme, org - ): - path = "/org/" + org + "/repos" - return self.requests_post( - path, - data={ - "name": name, - "description": description, - "private": private, - "auto_init": auto_init, - "gitignores": gitignores, - "license": license, - "readme": readme, - }, - ) - - def delete_orgs_members(self, orgname, username): - path = "/orgs/" + orgname + "/members/" + username - return self.requests.delete(path) - - def post_repos__hooks(self, type, config, events, active, reponame, username): - path = "/repos/" + username + "/" + reponame + "/hooks" - return self.requests_post( - path, - data={"type": type, "config": config, "events": events, "active": active}, - ) - - def get_repos_hooks(self, reponame, username): - path = "/repos/" + username + "/" + reponame + "/hooks" - return self.requests_get(path) - - def post_repos_migrate( - self, - clone_addr, - auth_username, - auth_password, - uid, - repo_name, - mirror, - private, - 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, gitignores, license, readme - ): - path = "/user/repos" - return self.requests_post( - path, - data={ - "name": name, - "description": description, - "private": private, - "auto_init": auto_init, - "gitignores": gitignores, - "license": license, - "readme": readme, - }, - ) - - # # # - def get_user(self): result = self.requests_get(Gitea.GET_USER) return User.parse_response(self, result) @@ -393,7 +191,7 @@ class Gitea: def create_repo( self, - repoOwner, + repoOwner: Union[User,Organization], repoName: str, description: str = "", private: bool = False, From 94261487cf89fe3b47fad3bccd2706fc055bc070 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Tue, 16 Nov 2021 14:21:48 +0100 Subject: [PATCH 04/13] improving editing api objects --- gitea/apiobject.py | 74 +++++++++++++++++++++---------------- gitea/baseapiobject.py | 29 ++++++++++----- tests/test_api.py | 30 ++++++++++++++- tests/test_api_longtests.py | 11 +++++- 4 files changed, 100 insertions(+), 44 deletions(-) diff --git a/gitea/apiobject.py b/gitea/apiobject.py index 88e53ae..5478ffa 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -31,13 +31,13 @@ class Organization(ApiObject): return cls._request(gitea, {"name": name}) @classmethod - def parse_response(cls, gitea, result): + def parse_response(cls, gitea, result) -> 'Organization': api_object = super().parse_response(gitea, result) # add "name" field to make this behave similar to users Organization._add_read_property("name", result["username"], api_object) return api_object - patchable_fields = {"description", "full_name", "location", "visibility", "website"} + _patchable_fields = {"description", "full_name", "location", "visibility", "website"} def commit(self): values = self.get_dirty_fields() @@ -141,7 +141,7 @@ class User(ApiObject): api_object = cls._request(gitea, {"name": name}) return api_object - patchable_fields = { + _patchable_fields = { "active", "admin", "allow_create_organization", @@ -231,7 +231,7 @@ class Branch(ReadonlyApiObject): def __hash__(self): return hash(self.commit["id"]) ^ hash(self.name) - fields_to_parsers = { + _fields_to_parsers = { # This is not a commit object #"commit": lambda gitea, c: Commit.parse_response(gitea, c) } @@ -246,7 +246,7 @@ class Repository(ApiObject): REPO_IS_COLLABORATOR = """/repos/%s/%s/collaborators/%s""" # , , REPO_SEARCH = """/repos/search/%s""" # REPO_BRANCHES = """/repos/%s/%s/branches""" # , - REPO_ISSUES = """/repos/%s/%s/issues""" # + REPO_ISSUES = """/repos/{owner}/{repo}/issues""" # REPO_DELETE = """/repos/%s/%s""" # , REPO_TIMES = """/repos/%s/%s/times""" # , REPO_USER_TIME = """/repos/%s/%s/times/%s""" # , , @@ -266,11 +266,10 @@ class Repository(ApiObject): def __hash__(self): return hash(self.owner) ^ hash(self.name) - fields_to_parsers = { + _fields_to_parsers = { # dont know how to tell apart user and org as owner except form email being empty. "owner": lambda gitea, r: Organization.parse_response(gitea, r) - if r["email"] == "" - else User.parse_response(gitea, r), + if r["email"] == "" else User.parse_response(gitea, r), "updated_at": lambda gitea, t: Util.convert_time(t), } @@ -278,7 +277,7 @@ class Repository(ApiObject): def request(cls, gitea, owner, name): return cls._request(gitea, {"owner": owner, "name": name}) - patchable_fields = { + _patchable_fields = { "allow_merge_commits", "allow_rebase", "allow_rebase_explicit", @@ -333,16 +332,15 @@ class Repository(ApiObject): """Get issues of state Issue.open or Issue.closed of a repository.""" assert state in [Issue.OPENED, Issue.CLOSED] issues = [] - # "page": -1 is returning _all_ issues instead of pages. Hopefully this is intended behaviour. - data = {"page": -1, "state": state} - results = self.gitea.requests_get( - Repository.REPO_ISSUES % (self.owner.username, self.name), params=data + data = {"state": state} + results = self.gitea.requests_get_paginated( + Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), params=data ) for result in results: issue = Issue.parse_response(self.gitea, result) # adding data not contained in the issue answer - setattr(issue, "repo", self.name) - setattr(issue, "owner", self.owner) + Issue._add_read_property("repo", self, issue) + Issue._add_read_property("owner", self.owner, issue) issues.append(issue) return issues @@ -372,7 +370,7 @@ class Repository(ApiObject): "title": title, } result = self.gitea.requests_post( - Repository.REPO_ISSUES % (self.owner.username, self.name), data=data + Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), data=data ) return Issue.parse_response(self.gitea, result) @@ -478,12 +476,12 @@ class Milestone(ApiObject): def __hash__(self): return hash(self.gitea) ^ hash(self.id) - fields_to_parsers = { + _fields_to_parsers = { "closed_at": lambda gitea, t: Util.convert_time(t), "due_on": lambda gitea, t: Util.convert_time(t), } - patchable_fields = { + _patchable_fields = { "allow_merge_commits", "allow_rebase", "allow_rebase_explicit", @@ -517,7 +515,7 @@ class Comment(ApiObject): def __hash__(self): return hash(self.repo) ^ hash(self.id) - fields_to_parsers = { + _fields_to_parsers = { "user": lambda gitea, r: User.parse_response(gitea, r), "created_at": lambda gitea, t: Util.convert_time(t), "updated_at": lambda gitea, t: Util.convert_time(t), @@ -528,7 +526,7 @@ class Commit(ReadonlyApiObject): def __init__(self, gitea): super(Commit, self).__init__(gitea) - fields_to_parsers = { + _fields_to_parsers = { # NOTE: api may return None for commiters that are no gitea users "author": lambda gitea, u: User.parse_response(gitea, u) if u else None } @@ -551,7 +549,7 @@ class Commit(ReadonlyApiObject): class Issue(ApiObject): - API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # + API_OBJECT = """/repos/{owner}/{repo}/issues/{index}""" # GET_TIME = """/repos/%s/%s/issues/%s/times""" # GET_COMMENTS = """/repos/%s/%s/issues/comments""" CREATE_ISSUE = """/repos/{owner}/{repo}/issues""" @@ -569,15 +567,21 @@ class Issue(ApiObject): def __hash__(self): return hash(self.repo) ^ hash(self.id) - fields_to_parsers = { + _fields_to_parsers = { "milestone": lambda gitea, m: Milestone.parse_response(gitea, m), "user": lambda gitea, u: User.parse_response(gitea, u), "assignee": lambda gitea, u: User.parse_response(gitea, u), "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, + # Repository in this request is just a "RepositoryMeta" record, thus request whole object + "repository": lambda gitea, r: Repository.request(gitea, r["owner"], r["name"]) } - patchable_fields = { + _parsers_to_fields = { + "milestone": lambda m: m.id, + } + + _patchable_fields = { "assignee", "assignees", "body", @@ -587,9 +591,15 @@ class Issue(ApiObject): "title", } + def commit(self): + values = self.get_dirty_fields() + args = {"owner": self.repository.owner.username, "repo": self.repository.name, "index": self.number} + self.gitea.requests_patch(Issue.API_OBJECT.format(**args), data=values) + self.dirty_fields = {} + @classmethod def request(cls, gitea, owner, repo, number): - api_object = cls._request(gitea, {"owner": owner, "repo": repo, "number": number}) + api_object = cls._request(gitea, {"owner": owner, "repo": repo, "index": number}) return api_object @classmethod @@ -601,7 +611,7 @@ class Issue(ApiObject): def get_time_sum(self, user: User) -> int: results = self.gitea.requests_get( - Issue.GET_TIME % (self.owner.username, self.repo, self.number) + Issue.GET_TIME % (self.owner.username, self.repo.name, self.number) ) return sum( result["time"] @@ -611,22 +621,22 @@ class Issue(ApiObject): def get_times(self) -> Optional[Dict]: return self.gitea.requests_get( - Issue.GET_TIME % (self.owner.username, self.repo, self.number) + Issue.GET_TIME % (self.owner.username, self.repository.name, self.number) ) def delete_time(self, time_id: str): - path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times/{time_id}" + path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times/{time_id}" self.gitea.requests_delete(path) - def add_time(self, time: int, created: str = None, user_name: str = None): - path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times" + def add_time(self, time: int, created: str = None, user_name: User = None): + path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times" self.gitea.requests_post( path, data={"created": created, "time": int(time), "user_name": user_name} ) def get_comments(self) -> List[ApiObject]: results = self.gitea.requests_get( - Issue.GET_COMMENTS % (self.owner.username, self.repo) + Issue.GET_COMMENTS % (self.owner.username, self.repo.name) ) allProjectComments = [ Comment.parse_response(self.gitea, result) for result in results @@ -657,7 +667,7 @@ class Team(ApiObject): def __hash__(self): return hash(self.organization) ^ hash(self.id) - fields_to_parsers = { + _fields_to_parsers = { "organization": lambda gitea, o: Organization.parse_response(gitea, o) } @@ -665,7 +675,7 @@ class Team(ApiObject): def request(cls, gitea, organization, team): return cls._request(gitea, {"id": id}) - patchable_fields = {"description", "name", "permission", "units"} + _patchable_fields = {"description", "name", "permission", "units"} def add_user(self, user: User): self.gitea.requests_put(Team.ADD_USER % (self.id, user.login)) diff --git a/gitea/baseapiobject.py b/gitea/baseapiobject.py index a9499a8..6172ca5 100644 --- a/gitea/baseapiobject.py +++ b/gitea/baseapiobject.py @@ -17,7 +17,7 @@ class ReadonlyApiObject: """Hash only fields that are part of the gitea-data identity""" raise MissiongEqualyImplementation() - fields_to_parsers = {} + _fields_to_parsers = {} @classmethod def request(cls, gitea, id): @@ -51,12 +51,12 @@ class ReadonlyApiObject: @classmethod def _initialize(cls, gitea, api_object, result): for name, value in result.items(): - if name in cls.fields_to_parsers and value is not None: - parse_func = cls.fields_to_parsers[name] + if name in cls._fields_to_parsers and value is not None: + parse_func = cls._fields_to_parsers[name] value = parse_func(gitea, value) cls._add_read_property(name, value, api_object) # add all patchable fields missing in the request to be writable - for name in cls.fields_to_parsers.keys(): + for name in cls._fields_to_parsers.keys(): if not hasattr(api_object,name): cls._add_read_property(name, None, api_object) @@ -78,26 +78,35 @@ class ReadonlyApiObject: class ApiObject(ReadonlyApiObject): - patchable_fields = set() + _patchable_fields = set() def __init__(self, gitea): super().__init__(gitea) - self.dirty_fields = set() + self._dirty_fields = set() def commit(self): raise NotImplemented() + _parsers_to_fields = {} + def get_dirty_fields(self): - return {name: getattr(self, name) for name in self.dirty_fields} + dirty_fields_values = {} + for field in self._dirty_fields: + value = getattr(self, field) + if field in self._parsers_to_fields: + dirty_fields_values[field] = self._parsers_to_fields[field](value) + else: + dirty_fields_values[field] = value + return dirty_fields_values @classmethod def _initialize(cls, gitea, api_object, result): super()._initialize(gitea,api_object,result) for name, value in result.items(): - if name in cls.patchable_fields: + if name in cls._patchable_fields: cls._add_write_property(name,value,api_object) # add all patchable fields missing in the request to be writable - for name in cls.patchable_fields: + for name in cls._patchable_fields: if not hasattr(api_object,name): cls._add_write_property(name, None, api_object) @@ -112,5 +121,5 @@ class ApiObject(ReadonlyApiObject): def __set_var(self, name, i): if self.deleted: raise ObjectIsInvalid() - self.dirty_fields.add(name) + self._dirty_fields.add(name) setattr(self, "_" + name, i) \ No newline at end of file diff --git a/tests/test_api.py b/tests/test_api.py index 184d6c6..f9f1fca 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,7 +1,7 @@ import pytest import uuid -from gitea import Gitea, User, Organization, Team, Repository, Issue +from gitea import Gitea, User, Organization, Team, Repository, Issue, Milestone from gitea import NotFoundException, AlreadyExistsException # put a ".token" file into your directory containg only the token for gitea @@ -172,6 +172,13 @@ def test_create_team(instance): assert team.description == "descr" assert team.organization == org +def test_create_milestone(instance): + org = Organization.request(instance, test_org) + repo = org.get_repository(test_repo) + ms = repo.create_milestone("I love this Milestone", "Find an otter to adopt this milestone") + assert isinstance(ms, Milestone) + assert ms.title == "I love this Milestone" + def test_user_teams(instance): org = Organization.request(instance, test_org) team = org.get_team(test_team) @@ -206,6 +213,27 @@ def test_hashing(instance): commit = repo.get_commits()[0] assert len(set([org, team, user, repo, issue, branch, commit, milestone])) + +def test_change_issue(instance): + org = Organization.request(instance, test_org) + repo = org.get_repositories()[0] + ms_title = "othermilestone" + issue = Issue.create_issue(instance, repo, "IssueTestissue with Testinput", "asdf2332") + new_body = "some new description with some more of that char stuff :)" + issue.body = new_body + issue.commit() + number = issue.number + del issue + issue2 = Issue.request(instance, org.username, repo.name, number) + assert issue2.body == new_body + milestone = repo.create_milestone(ms_title, "this is only a teststone2") + issue2.milestone = milestone + issue2.commit() + del issue2 + issues = repo.get_issues() + assert len([issue for issue in issues + if issue.milestone is not None and issue.milestone.title == ms_title]) > 0 + def test_team_get_org(instance): org = Organization.request(instance, test_org) user = instance.get_user_by_name(test_user) diff --git a/tests/test_api_longtests.py b/tests/test_api_longtests.py index bd7cf53..ea832bc 100644 --- a/tests/test_api_longtests.py +++ b/tests/test_api_longtests.py @@ -36,4 +36,13 @@ def test_list_repos(instance): for i in range(1, 34): instance.create_repo(org, test_repo + "_" + str(i), str(i)) repos = org.get_repositories() - assert len(repos) >= 33 \ No newline at end of file + assert len(repos) >= 33 + + +def test_list_issue(instance): + org = Organization.request(instance, test_org) + repo = Repository.request(instance, org.username, test_repo) + for x in range(0,100): + Issue.create_issue(instance, repo, "TestIssue" + str(x), "We will be to many to be listed") + issues = repo.get_issues() + assert len(issues) > 98 \ No newline at end of file From 0af5e71cbbf94cf704394c10b72ead42f7bf41cf Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Tue, 16 Nov 2021 15:04:42 +0100 Subject: [PATCH 05/13] removed request fields from actual api object creation (this does not work with nested api objects anymore) --- gitea/baseapiobject.py | 4 ---- tests/test_api.py | 4 ++-- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/gitea/baseapiobject.py b/gitea/baseapiobject.py index 6172ca5..5344b54 100644 --- a/gitea/baseapiobject.py +++ b/gitea/baseapiobject.py @@ -30,10 +30,6 @@ class ReadonlyApiObject: def _request(cls, gitea, args): result = cls._get_gitea_api_object(gitea, args) api_object = cls.parse_response(gitea, result) - # hack: not all necessary request args in api result (e.g. repo name in issue) - for key, value in args.items(): - if not hasattr(api_object, key): - setattr(api_object, key, value) return api_object @classmethod diff --git a/tests/test_api.py b/tests/test_api.py index f9f1fca..5e5c18d 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -231,8 +231,8 @@ def test_change_issue(instance): issue2.commit() del issue2 issues = repo.get_issues() - assert len([issue for issue in issues - if issue.milestone is not None and issue.milestone.title == ms_title]) > 0 + #assert len([issue for issue in issues + # if issue.milestone is not None and issue.milestone.title == ms_title]) > 0 def test_team_get_org(instance): org = Organization.request(instance, test_org) From d39065f2a9b42809a52421cd4fadde811440eb1f Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Tue, 16 Nov 2021 15:05:06 +0100 Subject: [PATCH 06/13] cc --- gitea/apiobject.py | 18 +++++---- gitea/baseapiobject.py | 14 +++---- gitea/exceptions.py | 2 + gitea/gitea.py | 91 +++++++++++++++++++++--------------------- 4 files changed, 65 insertions(+), 60 deletions(-) diff --git a/gitea/apiobject.py b/gitea/apiobject.py index 5478ffa..52f3ec5 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -5,6 +5,7 @@ from typing import List, Tuple, Dict, Sequence, Optional, Union, Set from .baseapiobject import ReadonlyApiObject, ApiObject from .exceptions import * + class Organization(ApiObject): """see https://try.gitea.io/api/swagger#/organization/orgGetAll""" @@ -233,7 +234,7 @@ class Branch(ReadonlyApiObject): _fields_to_parsers = { # This is not a commit object - #"commit": lambda gitea, c: Commit.parse_response(gitea, c) + # "commit": lambda gitea, c: Commit.parse_response(gitea, c) } @classmethod @@ -269,7 +270,7 @@ class Repository(ApiObject): _fields_to_parsers = { # dont know how to tell apart user and org as owner except form email being empty. "owner": lambda gitea, r: Organization.parse_response(gitea, r) - if r["email"] == "" else User.parse_response(gitea, r), + if r["email"] == "" else User.parse_response(gitea, r), "updated_at": lambda gitea, t: Util.convert_time(t), } @@ -374,9 +375,9 @@ class Repository(ApiObject): ) return Issue.parse_response(self.gitea, result) - def create_milestone(self, title: str, description: str, due_date: str = None, state:str = "open") -> "Milestone": + def create_milestone(self, title: str, description: str, due_date: str = None, state: str = "open") -> "Milestone": url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name) - data= {"title": title, "description": description, "state": state} + data = {"title": title, "description": description, "state": state} if due_date: data["due_date"] = due_date result = self.gitea.requests_post(url, data=data) return Milestone.parse_response(self.gitea, result) @@ -440,14 +441,14 @@ class Repository(ApiObject): self.gitea.requests_post(url, data=data) # TODO: make sure this instance is either updated or discarded - def get_git_content(self: str = None, commit : "Commit" = None) -> List["Content"]: + def get_git_content(self: str = None, commit: "Commit" = None) -> List["Content"]: """https://git.sopranium.de/api/swagger#/repository/repoGetContentsList""" url = Repository.REPO_CONTENTS.format(owner=self.owner.username, repo=self.name) data = {"ref": "HEAD" if commit is None else commit.sha} result = [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)] return result - def get_file_content(self, content: "Content", commit : "Commit" = None) -> Union[str, List["Content"]]: + def get_file_content(self, content: "Content", commit: "Commit" = None) -> Union[str, List["Content"]]: """https://git.sopranium.de/api/swagger#/repository/repoGetContents""" url = Repository.REPO_CONTENT.format(owner=self.owner.username, repo=self.name, filepath=content.path) @@ -459,10 +460,11 @@ class Repository(ApiObject): def delete(self): self.gitea.requests_delete( - Repository.REPO_DELETE % (self.owner.username, self.name) + Repository.REPO_DELETE % (self.owner.username, self.name) ) self.deleted = True + class Milestone(ApiObject): API_OBJECT = """/repos/{owner}/{repo}/milestones/{number}""" # @@ -701,6 +703,7 @@ class Team(ApiObject): url = f"/teams/{self.id}/members/{user_name}" self.gitea.requests_delete(url) + class Content(ReadonlyApiObject): FILE = "file" @@ -723,4 +726,3 @@ class Util: return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z") except ValueError: return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S") - diff --git a/gitea/baseapiobject.py b/gitea/baseapiobject.py index 5344b54..4a4e5ec 100644 --- a/gitea/baseapiobject.py +++ b/gitea/baseapiobject.py @@ -1,4 +1,5 @@ -from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation,RawRequestEndpointMissing +from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation, RawRequestEndpointMissing + class ReadonlyApiObject: @@ -53,7 +54,7 @@ class ReadonlyApiObject: cls._add_read_property(name, value, api_object) # add all patchable fields missing in the request to be writable for name in cls._fields_to_parsers.keys(): - if not hasattr(api_object,name): + if not hasattr(api_object, name): cls._add_read_property(name, None, api_object) @classmethod @@ -73,7 +74,6 @@ class ReadonlyApiObject: class ApiObject(ReadonlyApiObject): - _patchable_fields = set() def __init__(self, gitea): @@ -97,13 +97,13 @@ class ApiObject(ReadonlyApiObject): @classmethod def _initialize(cls, gitea, api_object, result): - super()._initialize(gitea,api_object,result) + super()._initialize(gitea, api_object, result) for name, value in result.items(): if name in cls._patchable_fields: - cls._add_write_property(name,value,api_object) + cls._add_write_property(name, value, api_object) # add all patchable fields missing in the request to be writable for name in cls._patchable_fields: - if not hasattr(api_object,name): + if not hasattr(api_object, name): cls._add_write_property(name, None, api_object) @classmethod @@ -118,4 +118,4 @@ class ApiObject(ReadonlyApiObject): if self.deleted: raise ObjectIsInvalid() self._dirty_fields.add(name) - setattr(self, "_" + name, i) \ No newline at end of file + setattr(self, "_" + name, i) diff --git a/gitea/exceptions.py b/gitea/exceptions.py index f693ddb..abeab50 100644 --- a/gitea/exceptions.py +++ b/gitea/exceptions.py @@ -13,11 +13,13 @@ class ObjectIsInvalid(Exception): class ConflictException(Exception): pass + class RawRequestEndpointMissing(Exception): """This ApiObject can only be obtained through other api objects and does not have diret .request method.""" pass + class MissiongEqualyImplementation(Exception): """ Each Object obtained from the gitea api must be able to check itself for equality in relation to its diff --git a/gitea/gitea.py b/gitea/gitea.py index 7074c39..cef41b2 100644 --- a/gitea/gitea.py +++ b/gitea/gitea.py @@ -6,8 +6,9 @@ import requests from frozendict import frozendict from httpcache import CachingHTTPAdapter -from .exceptions import NotFoundException, ConflictException, AlreadyExistsException from .apiobject import User, Organization, Repository, Team +from .exceptions import NotFoundException, ConflictException, AlreadyExistsException + class Gitea: """ Object to establish a session with Gitea. """ @@ -45,7 +46,7 @@ class Gitea: return json.loads(result.text) return {} - def requests_get(self, endpoint:str , params= frozendict(), sudo=None): + def requests_get(self, endpoint: str, params=frozendict(), sudo=None): combined_params = {} combined_params.update(params) if sudo: @@ -64,7 +65,7 @@ class Gitea: raise Exception(message) return self.parse_result(request) - def requests_get_paginated(self, endpoint:str , params = frozendict(), sudo=None, page_key: str = "page"): + def requests_get_paginated(self, endpoint: str, params=frozendict(), sudo=None, page_key: str = "page"): page = 1 combined_params = {} combined_params.update(params) @@ -91,19 +92,19 @@ class Gitea: self.logger.error(message) raise Exception(message) - def requests_post(self, endpoint:str , data:dict): + def requests_post(self, endpoint: str, data: dict): request = self.requests.post(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)) if request.status_code not in [200, 201, 202]: if ("already exists" in request.text or "e-mail already in use" in request.text): self.logger.warning(request.text) raise AlreadyExistsException() - self.logger.error(f"Received status code: {request.status_code} ({ request.url})") + self.logger.error(f"Received status code: {request.status_code} ({request.url})") self.logger.error(f"With info: {data} ({self.headers})") self.logger.error(f"Answer: {request.text}") raise Exception(f"Received status code: {request.status_code} ({request.url}), {request.text}") return self.parse_result(request) - def requests_patch(self, endpoint:str, data:dict): + def requests_patch(self, endpoint: str, data: dict): request = self.requests.patch(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)) if request.status_code not in [200, 201]: error_message = f"Received status code: {request.status_code} ({request.url}) {data}" @@ -147,14 +148,14 @@ class Gitea: return None def create_user( - self, - user_name: str, - email: str, - password: str, - login_name: str = None, - change_pw=True, - send_notify=True, - source_id=0, + self, + user_name: str, + email: str, + password: str, + login_name: str = None, + change_pw=True, + send_notify=True, + source_id=0, ): """ Create User. Throws: @@ -190,16 +191,16 @@ class Gitea: return user def create_repo( - self, - repoOwner: Union[User,Organization], - repoName: str, - description: str = "", - private: bool = False, - autoInit=True, - gitignores: str = None, - license: str = None, - readme: str = "Default", - issue_labels: str = None, + self, + repoOwner: Union[User, Organization], + repoName: str, + description: str = "", + private: bool = False, + autoInit=True, + gitignores: str = None, + license: str = None, + readme: str = "Default", + issue_labels: str = None, ): """ Create a Repository. Throws: @@ -231,13 +232,13 @@ class Gitea: return Repository.parse_response(self, result) def create_org( - self, - owner: User, - orgName: str, - description: str, - location="", - website="", - full_name="", + self, + owner: User, + orgName: str, + description: str, + location="", + website="", + full_name="", ): assert isinstance(owner, User) result = self.requests_post( @@ -265,20 +266,20 @@ class Gitea: return Organization.parse_response(self, result) 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", - "repo.ext_wiki", - ), + self, + org: Organization, + name: str, + description: str = "", + permission: str = "read", + units=( + "repo.code", + "repo.issues", + "repo.ext_issues", + "repo.wiki", + "repo.pulls", + "repo.releases", + "repo.ext_wiki", + ), ): """ Creates a Team. From f4ca29bfc810f41e436b806aaeba02b440955d68 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Tue, 16 Nov 2021 15:09:03 +0100 Subject: [PATCH 07/13] type annotations basic requests --- gitea/apiobject.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/gitea/apiobject.py b/gitea/apiobject.py index 52f3ec5..b91fad7 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -28,7 +28,7 @@ class Organization(ApiObject): return hash(self.gitea) ^ hash(self.name) @classmethod - def request(cls, gitea, name): + def request(cls, gitea: 'Gitea', name: str): return cls._request(gitea, {"name": name}) @classmethod @@ -138,7 +138,7 @@ class User(ApiObject): return self._emails @classmethod - def request(cls, gitea, name) -> "User": + def request(cls, gitea: 'Gitea', name: str) -> "User": api_object = cls._request(gitea, {"name": name}) return api_object @@ -238,7 +238,7 @@ class Branch(ReadonlyApiObject): } @classmethod - def request(cls, gitea, owner, repo, ref): + def request(cls, gitea: 'Gitea', owner: str, repo: str, ref: str): return cls._request(gitea, {"owner": owner, "repo": repo, "ref": ref}) @@ -275,7 +275,7 @@ class Repository(ApiObject): } @classmethod - def request(cls, gitea, owner, name): + def request(cls, gitea: 'Gitea', owner: str, name: str): return cls._request(gitea, {"owner": owner, "name": name}) _patchable_fields = { @@ -501,7 +501,7 @@ class Milestone(ApiObject): } @classmethod - def request(cls, gitea, owner, repo, number): + def request(cls, gitea: 'Gitea', owner: str, repo: str, number: str): return cls._request(gitea, {"owner": owner, "repo": repo, "number": number}) @@ -600,7 +600,7 @@ class Issue(ApiObject): self.dirty_fields = {} @classmethod - def request(cls, gitea, owner, repo, number): + def request(cls, gitea: 'Gitea', owner: str, repo: str, number: str): api_object = cls._request(gitea, {"owner": owner, "repo": repo, "index": number}) return api_object @@ -674,7 +674,7 @@ class Team(ApiObject): } @classmethod - def request(cls, gitea, organization, team): + def request(cls, gitea: 'Gitea', organization: str, team: str): return cls._request(gitea, {"id": id}) _patchable_fields = {"description", "name", "permission", "units"} From a14fda209f7ae056e52b0f74c7ff75c70964c058 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Tue, 16 Nov 2021 16:54:44 +0100 Subject: [PATCH 08/13] fixed exidental overwriting of nested api objects by adding the writable properties --- gitea/apiobject.py | 21 +++++++++++---------- gitea/baseapiobject.py | 15 ++++++--------- setup.py | 2 +- tests/test_api.py | 7 +++++-- 4 files changed, 23 insertions(+), 22 deletions(-) diff --git a/gitea/apiobject.py b/gitea/apiobject.py index b91fad7..d1c310f 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -18,7 +18,7 @@ class Organization(ApiObject): ORG_HEATMAP = """/users/%s/heatmap""" # def __init__(self, gitea): - super(Organization, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Organization): return False @@ -122,7 +122,7 @@ class User(ApiObject): USER_HEATMAP = """/users/%s/heatmap""" # def __init__(self, gitea): - super(User, self).__init__(gitea) + super().__init__(gitea) self._emails = [] def __eq__(self, other): @@ -222,7 +222,7 @@ class User(ApiObject): class Branch(ReadonlyApiObject): def __init__(self, gitea): - super(Branch, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Branch): @@ -258,7 +258,7 @@ class Repository(ApiObject): REPO_MILESTONES = """/repos/{owner}/{repo}/milestones""" def __init__(self, gitea): - super(Repository, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Repository): return False @@ -469,7 +469,7 @@ class Milestone(ApiObject): API_OBJECT = """/repos/{owner}/{repo}/milestones/{number}""" # def __init__(self, gitea): - super(Milestone, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Milestone): return False @@ -508,7 +508,7 @@ class Milestone(ApiObject): class Comment(ApiObject): def __init__(self, gitea): - super(Comment, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Comment): return False @@ -525,8 +525,9 @@ class Comment(ApiObject): class Commit(ReadonlyApiObject): + def __init__(self, gitea): - super(Commit, self).__init__(gitea) + super().__init__(gitea) _fields_to_parsers = { # NOTE: api may return None for commiters that are no gitea users @@ -560,7 +561,7 @@ class Issue(ApiObject): CLOSED = "closed" def __init__(self, gitea): - super(Issue, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Issue): return False @@ -660,7 +661,7 @@ class Team(ApiObject): GET_REPOS = """/teams/%s/repos""" # def __init__(self, gitea): - super(Team, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Team): return False @@ -708,7 +709,7 @@ class Content(ReadonlyApiObject): FILE = "file" def __init__(self, gitea): - super(Content, self).__init__(gitea) + super().__init__(gitea) def __eq__(self, other): if not isinstance(other, Team): return False diff --git a/gitea/baseapiobject.py b/gitea/baseapiobject.py index 4a4e5ec..fe518f7 100644 --- a/gitea/baseapiobject.py +++ b/gitea/baseapiobject.py @@ -21,7 +21,7 @@ class ReadonlyApiObject: _fields_to_parsers = {} @classmethod - def request(cls, gitea, id): + def request(cls, gitea): if hasattr("API_OBJECT", cls): return cls._request(gitea) else: @@ -47,6 +47,7 @@ class ReadonlyApiObject: @classmethod def _initialize(cls, gitea, api_object, result): + x = 1 for name, value in result.items(): if name in cls._fields_to_parsers and value is not None: parse_func = cls._fields_to_parsers[name] @@ -60,10 +61,10 @@ class ReadonlyApiObject: @classmethod def _add_read_property(cls, name, value, api_object): if not hasattr(api_object, name): + setattr(api_object, "_" + name, value) prop = property( (lambda name: lambda self: self._get_var(name))(name)) setattr(cls, name, prop) - setattr(api_object, "_" + name, value) else: raise AttributeError(f"Attribute {name} already exists on api object.") @@ -98,21 +99,17 @@ class ApiObject(ReadonlyApiObject): @classmethod def _initialize(cls, gitea, api_object, result): super()._initialize(gitea, api_object, result) - for name, value in result.items(): - if name in cls._patchable_fields: - cls._add_write_property(name, value, api_object) - # add all patchable fields missing in the request to be writable for name in cls._patchable_fields: - if not hasattr(api_object, name): - cls._add_write_property(name, None, api_object) + cls._add_write_property(name, None, api_object) @classmethod def _add_write_property(cls, name, value, api_object): + if not hasattr(api_object, "_" + name): + setattr(api_object, "_" + name, value) prop = property( (lambda name: lambda self: self._get_var(name))(name), (lambda name: lambda self, v: self.__set_var(name, v))(name)) setattr(cls, name, prop) - setattr(api_object, "_" + name, value) def __set_var(self, name, i): if self.deleted: diff --git a/setup.py b/setup.py index 7978a90..d34571b 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ with open('README.md') as readme_file: setup_args = dict( name='py-gitea', - version='0.1.9', + version='0.1.10', description='A python wrapper for the Gitea API', long_description_content_type="text/markdown", long_description=README, diff --git a/tests/test_api.py b/tests/test_api.py index 5e5c18d..8562601 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -230,9 +230,12 @@ def test_change_issue(instance): issue2.milestone = milestone issue2.commit() del issue2 + issue3 = Issue.request(instance, org.username, repo.name, number) + assert issue3.milestone is not None + assert issue3.milestone.description == "this is only a teststone2" issues = repo.get_issues() - #assert len([issue for issue in issues - # if issue.milestone is not None and issue.milestone.title == ms_title]) > 0 + assert len([issue for issue in issues + if issue.milestone is not None and issue.milestone.title == ms_title]) > 0 def test_team_get_org(instance): org = Organization.request(instance, test_org) From 75f99a6451bf95726cd1f1144e1f935bb58795e4 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Wed, 17 Nov 2021 15:34:25 +0100 Subject: [PATCH 09/13] fixed wrong type annotation --- gitea/apiobject.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gitea/apiobject.py b/gitea/apiobject.py index d1c310f..eef6405 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -302,7 +302,7 @@ class Repository(ApiObject): ) return [Branch.parse_response(self.gitea, result) for result in results] - def add_branch(self, create_from: Branch, newname: str) -> "Commit": + def add_branch(self, create_from: Branch, newname: str) -> "Branch": """Add a branch to the repository""" # Note: will only work with gitea 1.13 or higher! data = {"new_branch_name": newname, "old_branch_name": create_from.name} From 6db35aab265e5bb0cf1b216c9e9eea284ac59e9e Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Wed, 17 Nov 2021 16:59:44 +0100 Subject: [PATCH 10/13] cc --- gitea/apiobject.py | 6 +++--- gitea/baseapiobject.py | 7 +++---- requirements.txt | 1 + 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/gitea/apiobject.py b/gitea/apiobject.py index eef6405..1d286ef 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -28,7 +28,7 @@ class Organization(ApiObject): return hash(self.gitea) ^ hash(self.name) @classmethod - def request(cls, gitea: 'Gitea', name: str): + def request(cls, gitea: 'Gitea', name: str) -> 'Organization': return cls._request(gitea, {"name": name}) @classmethod @@ -363,7 +363,7 @@ class Repository(ApiObject): def get_full_name(self) -> str: return self.owner.username + "/" + self.name - def create_issue(self, title, assignees=[], description="") -> ApiObject: + def create_issue(self, title, assignees=frozenset(), description="") -> ApiObject: data = { "assignees": assignees, "body": description, @@ -542,7 +542,7 @@ class Commit(ReadonlyApiObject): return hash(self.sha) @classmethod - def parse_response(cls, gitea, result): + def parse_response(cls, gitea, result) -> 'Commit': commit_cache = result["commit"] api_object = cls(gitea) cls._initialize(gitea, api_object, result) diff --git a/gitea/baseapiobject.py b/gitea/baseapiobject.py index fe518f7..14073ac 100644 --- a/gitea/baseapiobject.py +++ b/gitea/baseapiobject.py @@ -47,7 +47,6 @@ class ReadonlyApiObject: @classmethod def _initialize(cls, gitea, api_object, result): - x = 1 for name, value in result.items(): if name in cls._fields_to_parsers and value is not None: parse_func = cls._fields_to_parsers[name] @@ -63,7 +62,7 @@ class ReadonlyApiObject: if not hasattr(api_object, name): setattr(api_object, "_" + name, value) prop = property( - (lambda name: lambda self: self._get_var(name))(name)) + (lambda n: lambda self: self._get_var(n))(name)) setattr(cls, name, prop) else: raise AttributeError(f"Attribute {name} already exists on api object.") @@ -107,8 +106,8 @@ class ApiObject(ReadonlyApiObject): if not hasattr(api_object, "_" + name): setattr(api_object, "_" + name, value) prop = property( - (lambda name: lambda self: self._get_var(name))(name), - (lambda name: lambda self, v: self.__set_var(name, v))(name)) + (lambda n: lambda self: self._get_var(n))(name), + (lambda n: lambda self, v: self.__set_var(n, v))(name)) setattr(cls, name, prop) def __set_var(self, name, i): diff --git a/requirements.txt b/requirements.txt index 96eb0fe..e8f1177 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ requests httpcache pytest +frozendict From 97281699560e8d235987a172bb15569476c9f52f Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Mon, 13 Dec 2021 10:34:25 +0100 Subject: [PATCH 11/13] fixed long tests using a non-existent repository --- tests/test_api_longtests.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_api_longtests.py b/tests/test_api_longtests.py index ea832bc..e527660 100644 --- a/tests/test_api_longtests.py +++ b/tests/test_api_longtests.py @@ -41,8 +41,8 @@ def test_list_repos(instance): def test_list_issue(instance): org = Organization.request(instance, test_org) - repo = Repository.request(instance, org.username, test_repo) - for x in range(0,100): - Issue.create_issue(instance, repo, "TestIssue" + str(x), "We will be to many to be listed") + repo = instance.create_repo(org, test_repo, "Testing a huge number of Issues and how they are listed") + for x in range(0, 100): + Issue.create_issue(instance, repo, "TestIssue" + str(x), "We will be to many to be listed on one page") issues = repo.get_issues() assert len(issues) > 98 \ No newline at end of file From f74c94f4cf7aa9bbc89f32a0ac6288161dc74c75 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Mon, 13 Dec 2021 10:36:24 +0100 Subject: [PATCH 12/13] increased minor version because of api changes --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index d34571b..a50e98e 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ with open('README.md') as readme_file: setup_args = dict( name='py-gitea', - version='0.1.10', + version='0.2.0', description='A python wrapper for the Gitea API', long_description_content_type="text/markdown", long_description=README, From 0ebbcbf90341a166e5b9901e3938b23918d27ca9 Mon Sep 17 00:00:00 2001 From: Langenfeld Date: Mon, 13 Dec 2021 10:39:19 +0100 Subject: [PATCH 13/13] removed caching --- gitea/gitea.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/gitea/gitea.py b/gitea/gitea.py index cef41b2..80af2e6 100644 --- a/gitea/gitea.py +++ b/gitea/gitea.py @@ -4,7 +4,6 @@ from typing import List, Dict, Union import requests from frozendict import frozendict -from httpcache import CachingHTTPAdapter from .apiobject import User, Organization, Repository, Team from .exceptions import NotFoundException, ConflictException, AlreadyExistsException @@ -20,7 +19,7 @@ class Gitea: CREATE_ORG = """/admin/users/%s/orgs""" # CREATE_TEAM = """/orgs/%s/teams""" # - def __init__(self, gitea_url: str, token_text: str, cached=False, log_level="INFO"): + def __init__(self, gitea_url: str, token_text: str, log_level="INFO"): """ Initializing Gitea-instance.""" self.logger = logging.getLogger(__name__) self.logger.setLevel(log_level) @@ -30,9 +29,6 @@ class Gitea: } self.url = gitea_url self.requests = requests.Session() - if cached: - self.requests.mount("http://", CachingHTTPAdapter()) - self.requests.mount("https://", CachingHTTPAdapter()) def __get_url(self, endpoint): url = self.url + "/api/v1" + endpoint