allspice.baseapiobject

  1from __future__ import annotations
  2
  3from typing import TYPE_CHECKING, Any, ClassVar, Mapping, Optional
  4
  5try:
  6    from typing_extensions import Self
  7except ImportError:
  8    from typing import Self
  9
 10if TYPE_CHECKING:
 11    from allspice.allspice import AllSpice
 12
 13from .exceptions import MissingEqualityImplementation, ObjectIsInvalid, RawRequestEndpointMissing
 14
 15
 16class ReadonlyApiObject:
 17    def __init__(self, allspice_client):
 18        self.allspice_client = allspice_client
 19        self.deleted = False  # set if .delete was called, so that an exception is risen
 20
 21    def __str__(self) -> str:
 22        return "AllSpiceObject (%s):" % (type(self))
 23
 24    def __eq__(self, other) -> bool:
 25        """Compare only fields that are part of the gitea-data identity"""
 26        raise MissingEqualityImplementation()
 27
 28    def __hash__(self) -> int:
 29        """Hash only fields that are part of the gitea-data identity"""
 30        raise MissingEqualityImplementation()
 31
 32    _fields_to_parsers: ClassVar[dict] = {}
 33
 34    # TODO: This should probably be made an abstract function as all children
 35    # redefine it.
 36    @classmethod
 37    def request(cls, allspice_client: AllSpice) -> Self:
 38        # This never would've worked, so maybe we should remove this function
 39        # outright.
 40        return cls._request(allspice_client)
 41
 42    @classmethod
 43    def _request(cls, allspice_client: AllSpice, args: Mapping) -> Self:
 44        result = cls._get_gitea_api_object(allspice_client, args)
 45        api_object = cls.parse_response(allspice_client, result)
 46        return api_object
 47
 48    @classmethod
 49    def _get_gitea_api_object(cls, allspice_client: AllSpice, args: Mapping) -> Mapping:
 50        """Retrieving an object always as GET_API_OBJECT"""
 51        if hasattr(cls, "API_OBJECT"):
 52            raw_request_endpoint = getattr(cls, "API_OBJECT")
 53            return allspice_client.requests_get(raw_request_endpoint.format(**args))
 54        else:
 55            raise RawRequestEndpointMissing()
 56
 57    @classmethod
 58    def parse_response(cls, allspice_client: AllSpice, result: Mapping) -> Self:
 59        # allspice_client.logger.debug("Found api object of type %s (id: %s)" % (type(cls), id))
 60        api_object = cls(allspice_client)
 61        cls._initialize(allspice_client, api_object, result)
 62        return api_object
 63
 64    @classmethod
 65    def _initialize(cls, allspice_client: AllSpice, api_object: Self, result: Mapping):
 66        for name, value in result.items():
 67            if name in cls._fields_to_parsers and value is not None:
 68                parse_func = cls._fields_to_parsers[name]
 69                value = parse_func(allspice_client, value)
 70            cls._add_read_property(name, value, api_object)
 71        # add all patchable fields missing in the request to be writable
 72        for name in cls._fields_to_parsers.keys():
 73            if not hasattr(api_object, name):
 74                cls._add_read_property(name, None, api_object)
 75
 76    @classmethod
 77    def _add_read_property(cls, name: str, value: Any, api_object: ReadonlyApiObject):
 78        if not hasattr(api_object, name):
 79            setattr(api_object, "_" + name, value)
 80            prop = property((lambda n: lambda self: self._get_var(n))(name))
 81            setattr(cls, name, prop)
 82        else:
 83            raise AttributeError(f"Attribute {name} already exists on api object.")
 84
 85    def _get_var(self, name: str) -> Any:
 86        if self.deleted:
 87            raise ObjectIsInvalid()
 88        return getattr(self, "_" + name)
 89
 90
 91class ApiObject(ReadonlyApiObject):
 92    _patchable_fields: ClassVar[set[str]] = set()
 93
 94    def __init__(self, allspice_client: AllSpice):
 95        super().__init__(allspice_client)
 96        self._dirty_fields = set()
 97
 98    def _commit(self, route_fields: dict, dirty_fields: Optional[Mapping] = None):
 99        if self.deleted:
100            raise ObjectIsInvalid()
101        if not hasattr(self, "API_OBJECT"):
102            raise RawRequestEndpointMissing()
103
104        raw_request_endpoint = getattr(self, "API_OBJECT")
105
106        if dirty_fields is None:
107            dirty_fields = self.get_dirty_fields()
108
109        self.allspice_client.requests_patch(
110            raw_request_endpoint.format(**route_fields),
111            dirty_fields,
112        )
113        self._dirty_fields = set()
114
115    def commit(self):
116        raise NotImplementedError()
117
118    _parsers_to_fields: ClassVar[dict] = {}
119
120    def get_dirty_fields(self) -> dict[str, Any]:
121        dirty_fields_values = {}
122        for field in self._dirty_fields:
123            value = getattr(self, field)
124            if field in self._parsers_to_fields:
125                dirty_fields_values[field] = self._parsers_to_fields[field](value)
126            else:
127                dirty_fields_values[field] = value
128        return dirty_fields_values
129
130    @classmethod
131    def _initialize(cls, allspice_client: AllSpice, api_object: Self, result: Mapping):
132        super()._initialize(allspice_client, api_object, result)
133        for name in cls._patchable_fields:
134            cls._add_write_property(name, None, api_object)
135
136    @classmethod
137    def _add_write_property(cls, name: str, value: Any, api_object: Self):
138        if not hasattr(api_object, "_" + name):
139            setattr(api_object, "_" + name, value)
140        prop = property(
141            (lambda n: lambda self: self._get_var(n))(name),
142            (lambda n: lambda self, v: self.__set_var(n, v))(name),
143        )
144        setattr(cls, name, prop)
145
146    def __set_var(self, name: str, value: Any):
147        if self.deleted:
148            raise ObjectIsInvalid()
149        self._dirty_fields.add(name)
150        setattr(self, "_" + name, value)