kopia lustrzana https://github.com/Langenfeld/py-gitea
cleaned up and renamed gitea api object bases
rodzic
05d906148a
commit
ad9dae0b0e
|
@ -1,14 +1,12 @@
|
||||||
from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation
|
from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation,RawRequestEndpointMissing
|
||||||
|
|
||||||
|
|
||||||
class BasicGiteaApiObject:
|
|
||||||
GET_API_OBJECT = "FORMAT/STINING/{argument}"
|
class ReadonlyApiObject:
|
||||||
PATCH_API_OBJECT = "FORMAT/STINING/{argument}"
|
|
||||||
|
|
||||||
def __init__(self, gitea):
|
def __init__(self, gitea):
|
||||||
self.gitea = gitea
|
self.gitea = gitea
|
||||||
self.deleted = False # set if .delete was called, so that an exception is risen
|
self.deleted = False # set if .delete was called, so that an exception is risen
|
||||||
self.dirty_fields = set()
|
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return "GiteaAPIObject (%s):" % (type(self))
|
return "GiteaAPIObject (%s):" % (type(self))
|
||||||
|
@ -21,71 +19,100 @@ class BasicGiteaApiObject:
|
||||||
"""Hash only fields that are part of the gitea-data identity"""
|
"""Hash only fields that are part of the gitea-data identity"""
|
||||||
raise MissiongEqualyImplementation()
|
raise MissiongEqualyImplementation()
|
||||||
|
|
||||||
|
|
||||||
fields_to_parsers = {}
|
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):
|
def commit(self):
|
||||||
raise NotImplemented()
|
raise NotImplemented()
|
||||||
|
|
||||||
def get_dirty_fields(self):
|
def get_dirty_fields(self):
|
||||||
return {name: getattr(self, name) for name in self.dirty_fields}
|
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
|
@classmethod
|
||||||
def _initialize(cls, gitea, api_object, result):
|
def _initialize(cls, gitea, api_object, result):
|
||||||
|
super()._initialize(gitea,api_object,result)
|
||||||
for name, value in result.items():
|
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:
|
if name in cls.patchable_fields:
|
||||||
cls._add_property(name, value, api_object)
|
cls._add_write_property(name,value,api_object)
|
||||||
else:
|
|
||||||
cls._add_readonly_property(name,value,api_object)
|
|
||||||
# add all patchable fields missing in the request to be writable
|
# 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):
|
if not hasattr(api_object,name):
|
||||||
cls._add_property(name, None, api_object)
|
cls._add_write_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)
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def _add_property(cls, name, value, api_object):
|
def _add_write_property(cls, name, value, api_object):
|
||||||
if not hasattr(api_object, name):
|
|
||||||
prop = property(
|
prop = property(
|
||||||
(lambda name: lambda self: self.__get_var(name))(name),
|
(lambda name: lambda self: self._get_var(name))(name),
|
||||||
(lambda name: lambda self, v: self.__set_var(name, v))(name))
|
(lambda name: lambda self, v: self.__set_var(name, v))(name))
|
||||||
setattr(cls, name, prop)
|
setattr(cls, name, prop)
|
||||||
setattr(api_object, "_" + name, value)
|
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 __set_var(self, name, i):
|
def __set_var(self, name, i):
|
||||||
if self.deleted:
|
if self.deleted:
|
||||||
raise ObjectIsInvalid()
|
raise ObjectIsInvalid()
|
||||||
self.dirty_fields.add(name)
|
self.dirty_fields.add(name)
|
||||||
setattr(self, "_" + name, i)
|
setattr(self, "_" + name, i)
|
||||||
|
|
||||||
def __get_var(self, name):
|
|
||||||
if self.deleted:
|
|
||||||
raise ObjectIsInvalid()
|
|
||||||
return getattr(self, "_" + name)
|
|
|
@ -13,6 +13,11 @@ class ObjectIsInvalid(Exception):
|
||||||
class ConflictException(Exception):
|
class ConflictException(Exception):
|
||||||
pass
|
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):
|
class MissiongEqualyImplementation(Exception):
|
||||||
"""
|
"""
|
||||||
Each Object obtained from the gitea api must be able to check itself for equality in relation to its
|
Each Object obtained from the gitea api must be able to check itself for equality in relation to its
|
||||||
|
|
|
@ -6,12 +6,11 @@ from typing import List, Tuple, Dict, Sequence, Optional, Union, Set
|
||||||
import requests
|
import requests
|
||||||
from httpcache import CachingHTTPAdapter
|
from httpcache import CachingHTTPAdapter
|
||||||
|
|
||||||
from .basicGiteaApiObject import BasicGiteaApiObject
|
from .baseapiobject import ReadonlyApiObject, ApiObject
|
||||||
from .exceptions import *
|
from .exceptions import *
|
||||||
from .giteaApiObject import GiteaApiObject
|
|
||||||
|
|
||||||
|
|
||||||
class Organization(GiteaApiObject):
|
class Organization(ApiObject):
|
||||||
"""see https://try.gitea.io/api/swagger#/organization/orgGetAll"""
|
"""see https://try.gitea.io/api/swagger#/organization/orgGetAll"""
|
||||||
|
|
||||||
GET_API_OBJECT = """/orgs/{name}""" # <org>
|
GET_API_OBJECT = """/orgs/{name}""" # <org>
|
||||||
|
@ -42,7 +41,7 @@ class Organization(GiteaApiObject):
|
||||||
def parse_response(cls, gitea, result):
|
def parse_response(cls, gitea, result):
|
||||||
api_object = super().parse_response(gitea, result)
|
api_object = super().parse_response(gitea, result)
|
||||||
# add "name" field to make this behave similar to users
|
# 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
|
return api_object
|
||||||
|
|
||||||
patchable_fields = {"description", "full_name", "location", "visibility", "website"}
|
patchable_fields = {"description", "full_name", "location", "visibility", "website"}
|
||||||
|
@ -100,7 +99,7 @@ class Organization(GiteaApiObject):
|
||||||
except:
|
except:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def remove_member(self, user: GiteaApiObject):
|
def remove_member(self, user: "User"):
|
||||||
path = f"/orgs/{self.username}/members/{user.username}"
|
path = f"/orgs/{self.username}/members/{user.username}"
|
||||||
self.gitea.requests_delete(path)
|
self.gitea.requests_delete(path)
|
||||||
|
|
||||||
|
@ -120,7 +119,7 @@ class Organization(GiteaApiObject):
|
||||||
return results
|
return results
|
||||||
|
|
||||||
|
|
||||||
class User(GiteaApiObject):
|
class User(ApiObject):
|
||||||
GET_API_OBJECT = """/users/{name}""" # <org>
|
GET_API_OBJECT = """/users/{name}""" # <org>
|
||||||
USER_MAIL = """/user/emails?sudo=%s""" # <name>
|
USER_MAIL = """/user/emails?sudo=%s""" # <name>
|
||||||
USER_PATCH = """/admin/users/%s""" # <username>
|
USER_PATCH = """/admin/users/%s""" # <username>
|
||||||
|
@ -226,18 +225,19 @@ class User(GiteaApiObject):
|
||||||
return results
|
return results
|
||||||
|
|
||||||
|
|
||||||
class Branch(GiteaApiObject):
|
class Branch(ReadonlyApiObject):
|
||||||
GET_API_OBJECT = """/repos/%s/%s/branches/%s""" # <owner>, <repo>, <ref>
|
GET_API_OBJECT = """/repos/%s/%s/branches/%s""" # <owner>, <repo>, <ref>
|
||||||
|
|
||||||
def __init__(self, gitea):
|
def __init__(self, gitea):
|
||||||
super(Branch, self).__init__(gitea)
|
super(Branch, self).__init__(gitea)
|
||||||
|
|
||||||
def __eq__(self, other):
|
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
|
return self.commit == other.commit and self.name == other.name
|
||||||
|
|
||||||
def __hash__(self):
|
def __hash__(self):
|
||||||
return hash(self.commit) ^ hash(self.name)
|
return hash(self.commit["id"]) ^ hash(self.name)
|
||||||
|
|
||||||
fields_to_parsers = {
|
fields_to_parsers = {
|
||||||
# This is not a commit object
|
# This is not a commit object
|
||||||
|
@ -249,7 +249,7 @@ class Branch(GiteaApiObject):
|
||||||
return cls._request(gitea, {"owner": owner, "repo": repo, "ref": ref})
|
return cls._request(gitea, {"owner": owner, "repo": repo, "ref": ref})
|
||||||
|
|
||||||
|
|
||||||
class Repository(GiteaApiObject):
|
class Repository(ApiObject):
|
||||||
REPO_IS_COLLABORATOR = (
|
REPO_IS_COLLABORATOR = (
|
||||||
"""/repos/%s/%s/collaborators/%s""" # <owner>, <reponame>, <username>
|
"""/repos/%s/%s/collaborators/%s""" # <owner>, <reponame>, <username>
|
||||||
)
|
)
|
||||||
|
@ -374,7 +374,7 @@ class Repository(GiteaApiObject):
|
||||||
def get_full_name(self) -> str:
|
def get_full_name(self) -> str:
|
||||||
return self.owner.username + "/" + self.name
|
return self.owner.username + "/" + self.name
|
||||||
|
|
||||||
def create_issue(self, title, assignees=[], description="") -> GiteaApiObject:
|
def create_issue(self, title, assignees=[], description="") -> ApiObject:
|
||||||
data = {
|
data = {
|
||||||
"assignees": assignees,
|
"assignees": assignees,
|
||||||
"body": description,
|
"body": description,
|
||||||
|
@ -475,7 +475,7 @@ class Repository(GiteaApiObject):
|
||||||
)
|
)
|
||||||
self.deleted = True
|
self.deleted = True
|
||||||
|
|
||||||
class Milestone(GiteaApiObject):
|
class Milestone(ApiObject):
|
||||||
GET_API_OBJECT = (
|
GET_API_OBJECT = (
|
||||||
"""/repos/{owner}/{repo}/milestones/{number}""" # <owner, repo>
|
"""/repos/{owner}/{repo}/milestones/{number}""" # <owner, repo>
|
||||||
)
|
)
|
||||||
|
@ -517,7 +517,7 @@ class Milestone(GiteaApiObject):
|
||||||
return cls._request(gitea, {"owner": owner, "repo": repo, "number": number})
|
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}"
|
PATCH_API_OBJECT = "/repos/{owner}/{repo}/issues/comments/{id}"
|
||||||
|
|
||||||
def __init__(self, gitea):
|
def __init__(self, gitea):
|
||||||
|
@ -539,7 +539,7 @@ class Comment(BasicGiteaApiObject):
|
||||||
patchable_fields = {"body"}
|
patchable_fields = {"body"}
|
||||||
|
|
||||||
|
|
||||||
class Commit(GiteaApiObject):
|
class Commit(ReadonlyApiObject):
|
||||||
def __init__(self, gitea):
|
def __init__(self, gitea):
|
||||||
super(Commit, self).__init__(gitea)
|
super(Commit, self).__init__(gitea)
|
||||||
|
|
||||||
|
@ -565,11 +565,11 @@ class Commit(GiteaApiObject):
|
||||||
commit_cache = result["commit"]
|
commit_cache = result["commit"]
|
||||||
api_object = cls(gitea)
|
api_object = cls(gitea)
|
||||||
cls._initialize(gitea, api_object, result)
|
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
|
return api_object
|
||||||
|
|
||||||
|
|
||||||
class Issue(GiteaApiObject):
|
class Issue(ApiObject):
|
||||||
GET_API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # <owner, repo, index>
|
GET_API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # <owner, repo, index>
|
||||||
GET_TIME = """/repos/%s/%s/issues/%s/times""" # <owner, repo, index>
|
GET_TIME = """/repos/%s/%s/issues/%s/times""" # <owner, repo, index>
|
||||||
GET_COMMENTS = """/repos/%s/%s/issues/comments"""
|
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}
|
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(
|
results = self.gitea.requests_get(
|
||||||
Issue.GET_COMMENTS % (self.owner.username, self.repo)
|
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}""" # <id>
|
GET_API_OBJECT = """/teams/{id}""" # <id>
|
||||||
ADD_USER = """/teams/%s/members/%s""" # <id, username to add>
|
ADD_USER = """/teams/%s/members/%s""" # <id, username to add>
|
||||||
ADD_REPO = """/teams/%s/repos/%s/%s""" # <id, org, repo>
|
ADD_REPO = """/teams/%s/repos/%s/%s""" # <id, org, repo>
|
||||||
|
@ -712,7 +712,7 @@ class Team(GiteaApiObject):
|
||||||
url = f"/teams/{self.id}/members/{user_name}"
|
url = f"/teams/{self.id}/members/{user_name}"
|
||||||
self.gitea.requests_delete(url)
|
self.gitea.requests_delete(url)
|
||||||
|
|
||||||
class Content(GiteaApiObject):
|
class Content(ReadonlyApiObject):
|
||||||
GET_API_OBJECT = """/repos/{owner}/{repo}/contents/{filepath}"""
|
GET_API_OBJECT = """/repos/{owner}/{repo}/contents/{filepath}"""
|
||||||
|
|
||||||
FILE = "file"
|
FILE = "file"
|
||||||
|
|
|
@ -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
|
|
||||||
|
|
||||||
|
|
Ładowanie…
Reference in New Issue