Allow to live update attributes without server restart

api/attributes/attributes.py

@@ -5,14 +5,13 @@
 
 from ayon_server.api.dependencies import AttributeName, CurrentUser
 from ayon_server.api.responses import EmptyResponse
-from ayon_server.api.system import require_server_restart
 from ayon_server.attributes.models import (
     AttributeModel,
     AttributePatchModel,
     AttributePutModel,
 )
 from ayon_server.attributes.validate_attribute_data import validate_attribute_data
-from ayon_server.entities import ProjectEntity
+from ayon_server.events import EventStream
 from ayon_server.exceptions import ForbiddenException, NotFoundException
 from ayon_server.lib.postgres import Postgres
 from ayon_server.types import OPModel
@@ -39,8 +38,7 @@ class SetAttributeListModel(GetAttributeListModel):
 async def save_attribute(attribute: AttributeModel) -> None:
     """Save attribute configuration to the database.
 
-    Additionally performs validation of the attribute data and updates
-    the enumerator in the running instance.
+    Additionally performs validation of the attribute data.
     """
     query = """
     INSERT INTO attributes
@@ -60,25 +58,6 @@ async def save_attribute(attribute: AttributeModel) -> None:
         attribute.data.dict(exclude_none=True),
     )
 
-    # TODO: The following code does not support horizontal scaling!!
-    # Notify other instances instead and reload the attribute library
-
-    if (enum := attribute.data.enum) is not None:
-        for name, field in ProjectEntity.model.attrib_model.__fields__.items():
-            if name != attribute.name:
-                continue
-
-            field_enum = field.field_info.extra.get("enum")
-            if field_enum is None:
-                continue
-            field_enum.clear()
-            field_enum.extend(enum)
-
-        for name, field in ProjectEntity.model.attrib_model.__fields__.items():
-            if name != attribute.name:
-                continue
-            field_enum = field.field_info.extra.get("enum")
-
 
 async def list_raw_attributes() -> list[dict[str, Any]]:
     """Return a list of attributes as they are stored in the DB"""
@@ -153,7 +132,7 @@ async def set_attribute_list(
     for attr in new_attributes:
         await save_attribute(attr)
 
-    await require_server_restart()
+    await EventStream.dispatch("server.attributes_updated")
     return EmptyResponse()
 
 
@@ -180,9 +159,7 @@ async def set_attribute_config(
         raise ForbiddenException("Only administrators are allowed to modify attributes")
     attribute = AttributeModel(name=attribute_name, **payload.dict())
     await save_attribute(attribute)
-    await require_server_restart(
-        None, "Restart the server to apply the attribute changes."
-    )
+    await EventStream.dispatch("server.attributes_updated")
     return EmptyResponse()
 
 
@@ -197,8 +174,6 @@ async def patch_attribute_config(
     patch_payload = payload.dict(exclude_unset=True)
     patch_data = patch_payload.pop("data", {})
 
-    requires_restart = False
-
     if "scope" in patch_payload or any(
         k in patch_data
         for k in (
@@ -216,8 +191,6 @@ async def patch_attribute_config(
             "inherit",
         )
     ):
-        requires_restart = True
-
         if not user.is_admin:
             raise ForbiddenException(
                 "Only administrators are allowed to modify attribute configuration"
@@ -236,10 +209,7 @@ async def patch_attribute_config(
 
     await save_attribute(attribute)
 
-    if requires_restart:
-        await require_server_restart(
-            None, "Restart the server to apply the attribute changes."
-        )
+    await EventStream.dispatch("server.attributes_updated")
     return EmptyResponse()
 
 
@@ -251,7 +221,5 @@ async def delete_attribute(
         raise ForbiddenException("Only administrators are allowed to delete attributes")
 
     await remove_attribute(attribute_name)
-    await require_server_restart(
-        None, "Restart the server to apply the attribute changes."
-    )
+    await EventStream.dispatch("server.attributes_updated")
     return EmptyResponse()

ayon_server/api/lifespan.py

@@ -8,6 +8,7 @@
 import semver
 
 from ayon_server.addons import AddonLibrary
+from ayon_server.entities.core import attribute_library
 from ayon_server.api.frontend import init_frontend
 from ayon_server.api.messaging import messaging
 from ayon_server.api.static import addon_static_router
@@ -110,6 +111,7 @@ async def lifespan(app: "FastAPI"):
         f.write(str(os.getpid()))
 
     await ayon_init()
+    EventStream.subscribe("server.attributes_updated", attribute_library.reload_handler, True)
     await load_access_groups()
     await CloudUtils.clear_cloud_info_cache()
 

ayon_server/entities/core/attrib.py

@@ -2,7 +2,7 @@
 import collections
 import functools
 import threading
-from typing import Any
+from typing import Any, Callable
 
 from ayon_server.lib.postgres import Postgres
 from ayon_server.logging import logger
@@ -29,6 +29,8 @@ def __init__(self) -> None:
         # in the same format as the attributes endpoint
         self.info_data: list[Any] = []
 
+        self._invalidation_callbacks: list[Callable[[], None]] = []
+
         # We need to load attribute data in a separate thread
         # with a separate event loop, because the main event loop
         # is already running and we cannot run another one
@@ -146,5 +148,58 @@ def by_name_scoped(self, entity_type: str, name: str) -> dict[str, Any]:
                 return attr
         raise KeyError(f"Attribute {name} not found for entity type {entity_type}")
 
+    def register_invalidation_callback(self, callback: Callable[[], None]) -> None:
+        """Register a callback to be called when attributes are reloaded."""
+        self._invalidation_callbacks.append(callback)
+
+    async def reload(self) -> None:
+        """Reload attributes from the database and invalidate all cached models.
+
+        Fetches fresh data from the database, updates the in-memory attribute
+        lists in-place (preserving list object identity so ModelSet.attributes
+        references remain valid), and invalidates all cached Pydantic models
+        so they are regenerated on next access.
+        """
+        query = "SELECT * FROM public.attributes ORDER BY position"
+        result = await Postgres.fetch(query)
+
+        # Build new data in a temporary structure first to minimize
+        # the window of inconsistency
+        new_data: collections.defaultdict[str, list[Any]] = collections.defaultdict(list)
+        new_info_data: list[Any] = []
+
+        for row in result:
+            new_info_data.append(row)
+            for scope in row["scope"]:
+                attrd = {"name": row["name"], **row["data"]}
+                if (scope != "project") and ("default" in attrd):
+                    del attrd["default"]
+                new_data[scope].append(attrd)
+
+        # Update self.data in-place to preserve list object identity.
+        # ModelSet instances hold direct references to these list objects,
+        # so we must mutate them rather than replace them.
+        all_scopes = set(self.data.keys()) | set(new_data.keys())
+        for scope in all_scopes:
+            self.data[scope].clear()
+            self.data[scope].extend(new_data.get(scope, []))
+
+        self.info_data = new_info_data
+
+        # Clear functools caches since the underlying data has changed
+        AttributeLibrary.inheritable_attributes.cache_clear()
+        AttributeLibrary.by_name.cache_clear()
+        AttributeLibrary.by_name_scoped.cache_clear()
+
+        # Invalidate all registered ModelSet Pydantic model caches
+        for callback in self._invalidation_callbacks:
+            callback()
+
+        logger.info("Attribute library reloaded")
+
+    async def reload_handler(self, event: Any = None) -> None:
+        """Event handler adapter for reload(), for use with EventStream.subscribe."""
+        await self.reload()
+
 
 attribute_library = AttributeLibrary()

ayon_server/entities/models/__init__.py

@@ -16,6 +16,7 @@
     version_fields,
     workfile_fields,
 )
+from ayon_server.entities.core.attrib import attribute_library
 from ayon_server.entities.models.generator import generate_model
 from ayon_server.types import (
     ENTITY_ID_EXAMPLE,
@@ -85,6 +86,19 @@ def __init__(
         self._patch_model: type[BaseModel] | None = None
         self._attrib_model: type[BaseModel] | None = None
 
+        attribute_library.register_invalidation_callback(self.invalidate)
+
+    def invalidate(self) -> None:
+        """Invalidate all cached Pydantic models.
+
+        Forces regeneration of all cached models on next access. Called
+        by AttributeLibrary.reload() when attributes are updated live.
+        """
+        self._attrib_model = None
+        self._model = None
+        self._post_model = None
+        self._patch_model = None
+
     @property
     def attrib_model(self) -> type[BaseModel]:
         """Return the attribute model."""

ayon_server/events/default_hooks.py

@@ -2,6 +2,8 @@
 
 from typing import TYPE_CHECKING
 
+import aiocache
+
 from ayon_server.lib.redis import Redis
 from ayon_server.logging import logger
 
@@ -16,8 +18,19 @@ async def clear_settings_cache(event: "EventModel"):
     await Redis.delete_ns("all-settings")
 
 
+async def clear_attribute_info_cache(event: "EventModel"):
+    """Clear the in-memory aiocache for the /info attributes response.
+
+    Called on all nodes via global hook so each instance flushes its own
+    local cache immediately when attributes are updated.
+    """
+    logger.trace("Clearing attribute info cache")
+    await aiocache.caches.get("default").clear()
+
+
 DEFAULT_HOOKS: list[tuple[str, HandlerType, bool]] = [
     ("settings.changed", clear_settings_cache, False),
     ("bundle.created", clear_settings_cache, False),
     ("bundle.updated", clear_settings_cache, False),
+    ("server.attributes_updated", clear_attribute_info_cache, True),
 ]