kopia lustrzana https://github.com/Langenfeld/py-gitea
121 wiersze
3.9 KiB
Python
121 wiersze
3.9 KiB
Python
from .exceptions import (
|
|
ObjectIsInvalid,
|
|
MissiongEqualyImplementation,
|
|
RawRequestEndpointMissing,
|
|
)
|
|
|
|
|
|
class ReadonlyApiObject:
|
|
def __init__(self, gitea):
|
|
self.gitea = gitea
|
|
self.deleted = False # set if .delete was called, so that an exception is risen
|
|
|
|
def __str__(self):
|
|
return "GiteaAPIObject (%s):" % (type(self))
|
|
|
|
def __eq__(self, other):
|
|
"""Compare only fields that are part of the gitea-data identity"""
|
|
raise MissiongEqualyImplementation()
|
|
|
|
def __hash__(self):
|
|
"""Hash only fields that are part of the gitea-data identity"""
|
|
raise MissiongEqualyImplementation()
|
|
|
|
_fields_to_parsers = {}
|
|
|
|
@classmethod
|
|
def request(cls, gitea):
|
|
if hasattr("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)
|
|
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.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):
|
|
setattr(api_object, "_" + name, value)
|
|
prop = property((lambda n: lambda self: self._get_var(n))(name))
|
|
setattr(cls, name, prop)
|
|
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()
|
|
|
|
_parsers_to_fields = {}
|
|
|
|
def get_dirty_fields(self):
|
|
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 in cls._patchable_fields:
|
|
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 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):
|
|
if self.deleted:
|
|
raise ObjectIsInvalid()
|
|
self._dirty_fields.add(name)
|
|
setattr(self, "_" + name, i)
|