better handling of cached attributes

Author: iLLiCiTiT

ayon_api/_api_helpers/attributes.py

@@ -1,38 +1,107 @@
 from __future__ import annotations
 
+import copy
+import time
 import typing
 from typing import Optional
-import copy
 
 from .base import BaseServerAPI
 
 if typing.TYPE_CHECKING:
     from ayon_api.typing import (
+        AttributeSchemaDict,
         AttributeSchemaDataDict,
         AttributesSchemaDict,
         AttributeScope,
     )
 
+class _AttributesCache:
+    _schema = None
+    _last_fetch = 0
+    _timeout = 60
+    _attributes_by_type = {}
+
+    def reset_schema(self) -> None:
+        self._schema = None
+        self._last_fetch = 0
+        self._attributes_by_type = {}
+
+    def set_timeout(self, timeout: int) -> None:
+        self._timeout = timeout
+
+    def get_schema(self) -> AttributesSchemaDict:
+        return copy.deepcopy(self._schema)
+
+    def set_schema(self, schema: AttributesSchemaDict) -> None:
+        self._schema = schema
+        self._last_fetch = time.time()
+
+    def is_valid(self) -> bool:
+        if self._schema is None:
+            return False
+        return time.time() - self._last_fetch < self._timeout
+
+    def invalidate(self) -> None:
+        if not self.is_valid():
+            self.reset_schema()
+
+    def get_attributes_for_type(
+        self, entity_type: AttributeScope
+    ) -> list[AttributeSchemaDict]:
+        attributes = self._attributes_by_type.get(entity_type)
+        if attributes is not None:
+            return attributes
+
+        attributes_schema = self.get_schema()
+        if attributes_schema is None:
+            raise ValueError("Attributes schema is not cached.")
+
+        attributes = []
+        for attr in attributes_schema["attributes"]:
+            if entity_type not in attr["scope"]:
+                continue
+            attributes.append(attr)
+
+        self._attributes_by_type[entity_type] = attributes
+        return attributes
+
+
 
 class AttributesAPI(BaseServerAPI):
-    _attributes_schema = None
-    _entity_type_attributes_cache = {}
+    _attributes_cache = _AttributesCache()
 
     def get_attributes_schema(
         self, use_cache: bool = True
     ) -> AttributesSchemaDict:
         if not use_cache:
-            self.reset_attributes_schema()
+            self._attributes_cache.reset_schema()
+        else:
+            self._attributes_cache.invalidate()
 
-        if self._attributes_schema is None:
+        if not self._attributes_cache.is_valid():
             result = self.get("attributes")
             result.raise_for_status()
-            self._attributes_schema = result.data
-        return copy.deepcopy(self._attributes_schema)
+            self._attributes_cache.set_schema(result.data)
+        return self._attributes_cache.get_schema()
 
     def reset_attributes_schema(self) -> None:
-        self._attributes_schema = None
-        self._entity_type_attributes_cache = {}
+        """Reset attributes schema cache.
+
+        DEPRECATED:
+            Use 'reset_attributes_cache' instead.
+
+        """
+        self.log.warning(
+            "Used deprecated function 'reset_attributes_schema'."
+            " Please use 'reset_attributes_cache' instead."
+        )
+        self.reset_attributes_cache()
+
+    def reset_attributes_cache(self) -> None:
+        self._attributes_cache.reset_schema()
+
+    def set_attributes_cache_timeout(self, timeout: int) -> None:
+        self._attributes_cache.set_timeout(timeout)
 
     def set_attribute_config(
         self,
@@ -63,12 +132,10 @@ def set_attribute_config(
             position=position,
             builtin=builtin
         )
-        if response.status_code != 204:
-            # TODO raise different exception
-            raise ValueError(
-                f"Attribute \"{attribute_name}\" was not created/updated."
-                f" {response.detail}"
-            )
+        response.raise_for_status(
+            f"Attribute \"{attribute_name}\" was not created/updated."
+            f" {response.detail}"
+        )
 
         self.reset_attributes_schema()
 
@@ -128,19 +195,14 @@ def get_attributes_for_type(
                 for entered entity type.
 
         """
-        attributes = self._entity_type_attributes_cache.get(entity_type)
-        if attributes is None:
-            attributes_schema = self.get_attributes_schema()
-            attributes = {}
-            for attr in attributes_schema["attributes"]:
-                if entity_type not in attr["scope"]:
-                    continue
-                attr_name = attr["name"]
-                attributes[attr_name] = attr["data"]
-
-            self._entity_type_attributes_cache[entity_type] = attributes
-
-        return copy.deepcopy(attributes)
+        # Make sure attributes are cached
+        self.get_attributes_schema()
+        return {
+            attr["name"]: attr["data"]
+            for attr in self._attributes_cache.get_attributes_for_type(
+                entity_type
+            )
+        }
 
     def get_attributes_fields_for_type(
         self, entity_type: AttributeScope