node-level create and read user timestamps

backend/infrahub/core/attribute.py

@@ -334,7 +334,7 @@ async def save(self, db: InfrahubDatabase, at: Timestamp | None = None) -> Attri
         if not self.id:
             return None
 
-        return await self._update(at=save_at, db=db)
+        return await self._update(db=db, at=save_at)
 
     async def delete(self, db: InfrahubDatabase, at: Timestamp | None = None) -> AttributeChangelog | None:
         if not self.db_id:

backend/infrahub/core/constants/__init__.py

@@ -46,6 +46,8 @@
 
 EVENT_NAMESPACE = "infrahub"
 
+SYSTEM_USER_ID = "__system__"
+
 
 class EventType(InfrahubStringEnum):
     BRANCH_CREATED = f"{EVENT_NAMESPACE}.branch.created"
@@ -371,6 +373,7 @@ class MetadataOptions(Flag):
     UPDATED_AT = auto()
     TIMESTAMPS = CREATED_AT | UPDATED_AT
     USERS = CREATED_BY | UPDATED_BY
+    USER_TIMESTAMPS = TIMESTAMPS | USERS
 
 
 RESTRICTED_NAMESPACES: list[str] = [

backend/infrahub/core/manager.py

@@ -1156,7 +1156,7 @@ async def get_many(
 
         # Query all nodes
         query = await NodeListGetInfoQuery.init(
-            db=db, ids=ids, branch=branch, account=account, at=at, branch_agnostic=branch_agnostic
+            db=db, ids=ids, branch=branch, at=at, branch_agnostic=branch_agnostic, include_metadata=include_metadata
         )
         await query.execute(db=db)
         nodes_info_by_id: dict[str, NodeToProcess] = {node.node_uuid: node async for node in query.get_nodes(db=db)}
@@ -1185,7 +1185,6 @@ async def get_many(
             new_node_data: dict[str, str | AttributeFromDB] = {
                 "db_id": node.node_id,
                 "id": node_id,
-                "updated_at": node.updated_at,
             }
 
             if not node.schema:
@@ -1206,6 +1205,10 @@ async def get_many(
             node_branch = await registry.get_branch(db=db, branch=node.branch)
             item = await node_class.init(schema=node.schema, branch=node_branch, at=at, db=db)
             await item.load(**new_node_data, db=db)
+            item._set_created_at(node.created_at)
+            item._set_created_by(node.created_by)
+            item._set_updated_at(node.updated_at)
+            item._set_updated_by(node.updated_by)
 
             nodes[node_id] = item
 

backend/infrahub/core/metadata/interface.py

@@ -0,0 +1,37 @@
+from abc import abstractmethod
+
+from infrahub.core.timestamp import Timestamp
+
+
+class MetadataInterface:
+    @abstractmethod
+    def _set_created_at(self, value: Timestamp | None) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _set_created_by(self, value: str | None) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _set_updated_at(self, value: Timestamp | None) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _set_updated_by(self, value: str | None) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _get_created_at(self) -> Timestamp | None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _get_created_by(self) -> str | None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _get_updated_at(self) -> Timestamp | None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _get_updated_by(self) -> str | None:
+        raise NotImplementedError()

backend/infrahub/core/metadata/model.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+from infrahub.core.timestamp import Timestamp
+
+
+@dataclass
+class MetadataInfo:
+    created_at: Timestamp | None = None
+    created_by: str | None = None
+    updated_at: Timestamp | None = None
+    updated_by: str | None = None

backend/infrahub/core/node/__init__.py

@@ -14,6 +14,7 @@
     GLOBAL_BRANCH_NAME,
     OBJECT_TEMPLATE_NAME_ATTR,
     OBJECT_TEMPLATE_RELATIONSHIP_NAME,
+    SYSTEM_USER_ID,
     BranchSupportType,
     ComputedAttributeKind,
     InfrahubKind,
@@ -22,8 +23,10 @@
     RelationshipKind,
 )
 from infrahub.core.constants.schema import SchemaElementPathType
+from infrahub.core.metadata.interface import MetadataInterface
+from infrahub.core.metadata.model import MetadataInfo
 from infrahub.core.protocols import CoreNumberPool, CoreObjectTemplate
-from infrahub.core.query.node import NodeCheckIDQuery, NodeCreateAllQuery, NodeDeleteQuery, NodeGetListQuery
+from infrahub.core.query.node import NodeCheckIDQuery, NodeCreateAllQuery, NodeDeleteQuery
 from infrahub.core.schema import (
     AttributeSchema,
     GenericSchema,
@@ -40,12 +43,10 @@
 from infrahub.types import ATTRIBUTE_TYPES
 
 from ...graphql.constants import KIND_GRAPHQL_FIELD_NAME
-from ...graphql.models import OrderModel
 from ...log import get_logger
 from ..attribute import BaseAttribute
 from ..query.relationship import RelationshipDeleteAllQuery
 from ..relationship import RelationshipManager
-from ..utils import update_relationships_to
 from .base import BaseNode, BaseNodeMeta, BaseNodeOptions
 from .node_property_attribute import DisplayLabel, HumanFriendlyIdentifier
 
@@ -69,7 +70,7 @@
 log = get_logger()
 
 
-class Node(BaseNode, metaclass=BaseNodeMeta):
+class Node(BaseNode, MetadataInterface, metaclass=BaseNodeMeta):
     @classmethod
     def __init_subclass_with_meta__(
         cls, _meta: BaseNodeOptions | None = None, default_filter: None = None, **options: dict[str, Any]
@@ -81,12 +82,13 @@ def __init_subclass_with_meta__(
         super().__init_subclass_with_meta__(_meta=_meta, **options)
 
     def __init__(self, schema: NodeSchema | ProfileSchema | TemplateSchema, branch: Branch, at: Timestamp):
+        super().__init__()
         self._schema: NodeSchema | ProfileSchema | TemplateSchema = schema
         self._branch: Branch = branch
         self._at: Timestamp = at
         self._existing: bool = False
+        self._metadata = MetadataInfo()
 
-        self._updated_at: Timestamp | None = None
         self.id: str = None
         self.db_id: str = None
 
@@ -103,6 +105,30 @@ def __init__(self, schema: NodeSchema | ProfileSchema | TemplateSchema, branch:
         self._relationships: list[str] = []
         self._node_changelog: NodeChangelog | None = None
 
+    def _set_created_at(self, value: Timestamp | None) -> None:
+        self._metadata.created_at = value
+
+    def _set_created_by(self, value: str | None) -> None:
+        self._metadata.created_by = value
+
+    def _set_updated_at(self, value: Timestamp | None) -> None:
+        self._metadata.updated_at = value
+
+    def _set_updated_by(self, value: str | None) -> None:
+        self._metadata.updated_by = value
+
+    def _get_created_at(self) -> Timestamp | None:
+        return self._metadata.created_at
+
+    def _get_created_by(self) -> str | None:
+        return self._metadata.created_by
+
+    def _get_updated_at(self) -> Timestamp | None:
+        return self._metadata.updated_at
+
+    def _get_updated_by(self) -> str | None:
+        return self._metadata.updated_by
+
     def get_schema(self) -> NonGenericSchemaTypes:
         return self._schema
 
@@ -120,9 +146,6 @@ def get_id(self) -> str:
 
         raise InitializationError("The node has not been saved yet and doesn't have an id")
 
-    def get_updated_at(self) -> Timestamp | None:
-        return self._updated_at
-
     def get_attribute(self, name: str) -> BaseAttribute:
         attribute = getattr(self, name)
         if not isinstance(attribute, BaseAttribute):
@@ -786,19 +809,12 @@ async def load(
         db: InfrahubDatabase,
         id: str | None = None,
         db_id: str | None = None,
-        updated_at: Timestamp | str | None = None,
         **kwargs: Any,
     ) -> Self:
         self.id = id
         self.db_id = db_id
         self._existing = True
 
-        if updated_at:
-            kwargs["updated_at"] = (
-                updated_at  # FIXME: Allow users to use "updated_at" named attributes until we have proper metadata handling
-            )
-            self._updated_at = Timestamp(updated_at)
-
         if not self._schema.is_schema_node:
             if hfid := kwargs.pop("human_friendly_id", None):
                 self._human_friendly_id = HumanFriendlyIdentifier(
@@ -812,19 +828,22 @@ async def load(
         await self._process_fields(db=db, fields=kwargs)
         return self
 
-    async def _create(self, db: InfrahubDatabase, at: Timestamp | None = None) -> NodeChangelog:
+    async def _create(self, db: InfrahubDatabase, user_id: str, at: Timestamp | None = None) -> NodeChangelog:
         create_at = Timestamp(at)
+        self._set_created_at(create_at)
+        self._set_created_by(user_id)
+        self._set_updated_at(create_at)
+        self._set_updated_by(user_id)
 
         if not self._schema.is_schema_node:
             await self.add_human_friendly_id(db=db)
             await self.add_display_label(db=db)
 
-        query = await NodeCreateAllQuery.init(db=db, node=self, at=create_at)
+        query = await NodeCreateAllQuery.init(db=db, node=self, at=create_at, user_id=user_id)
         await query.execute(db=db)
 
         _, self.db_id = query.get_self_ids()
         self._at = create_at
-        self._updated_at = create_at
         self._existing = True
 
         new_ids = query.get_ids()
@@ -856,7 +875,7 @@ async def _create(self, db: InfrahubDatabase, at: Timestamp | None = None) -> No
         return node_changelog
 
     async def _update(
-        self, db: InfrahubDatabase, at: Timestamp | None = None, fields: list[str] | None = None
+        self, db: InfrahubDatabase, user_id: str, at: Timestamp | None = None, fields: list[str] | None = None
     ) -> NodeChangelog:
         """Update the node in the database if needed."""
 
@@ -866,8 +885,8 @@ async def _update(
         # Go over the list of Attribute and update them one by one
         for name in self._attributes:
             if (fields and name in fields) or not fields:
-                attr: BaseAttribute = getattr(self, name)
-                updated_attribute = await attr.save(at=update_at, db=db)
+                attr = self.get_attribute(name=name)
+                updated_attribute = await attr.save(db=db, at=update_at)
                 if updated_attribute:
                     node_changelog.add_attribute(attribute=updated_attribute)
 
@@ -876,15 +895,15 @@ async def _update(
         for name in self._relationships:
             if (fields and name in fields) or not fields:
                 processed_relationships.append(name)
-                rel: RelationshipManager = getattr(self, name)
-                updated_relationship = await rel.save(at=update_at, db=db)
+                rel = self.get_relationship(name=name)
+                updated_relationship = await rel.save(db=db, at=update_at)
                 node_changelog.add_relationship(relationship_changelog=updated_relationship)
 
         if len(processed_relationships) != len(self._relationships):
             # Analyze if the node has a parent and add it to the changelog if missing
             if parent_relationship := self._get_parent_relationship_name():
                 if parent_relationship not in processed_relationships:
-                    rel: RelationshipManager = getattr(self, parent_relationship)
+                    rel = self.get_relationship(name=parent_relationship)
                     if parent := await rel.get_parent(db=db):
                         node_changelog.add_parent_from_relationship(parent=parent)
 
@@ -912,20 +931,31 @@ async def _update(
                 node_changelog.add_attribute(attribute=updated_attribute)
 
         node_changelog.display_label = await self.get_display_label(db=db)
+
+        if node_changelog.has_changes:
+            self._set_updated_at(update_at)
+            self._set_updated_by(user_id)
+
         return node_changelog
 
-    async def save(self, db: InfrahubDatabase, at: Timestamp | None = None, fields: list[str] | None = None) -> Self:
+    async def save(
+        self,
+        db: InfrahubDatabase,
+        user_id: str = SYSTEM_USER_ID,
+        at: Timestamp | None = None,
+        fields: list[str] | None = None,
+    ) -> Self:
         """Create or Update the Node in the database."""
         save_at = Timestamp(at)
 
         if self._existing:
-            self._node_changelog = await self._update(at=save_at, db=db, fields=fields)
+            self._node_changelog = await self._update(db=db, user_id=user_id, at=save_at, fields=fields)
         else:
-            self._node_changelog = await self._create(at=save_at, db=db)
+            self._node_changelog = await self._create(db=db, user_id=user_id, at=save_at)
 
         return self
 
-    async def delete(self, db: InfrahubDatabase, at: Timestamp | None = None) -> None:
+    async def delete(self, db: InfrahubDatabase, user_id: str = SYSTEM_USER_ID, at: Timestamp | None = None) -> None:
         """Delete the Node in the database."""
 
         delete_at = Timestamp(at)
@@ -962,22 +992,7 @@ async def delete(self, db: InfrahubDatabase, at: Timestamp | None = None) -> Non
         for relationship_changelog in deleted_relationships_changelogs:
             node_changelog.add_relationship(relationship_changelog=relationship_changelog)
 
-        # Update the relationship to the branch itself
-        query = await NodeGetListQuery.init(
-            db=db,
-            schema=self._schema,
-            filters={"id": self.id},
-            branch=self._branch,
-            at=delete_at,
-            order=OrderModel(disable=True),
-        )
-        await query.execute(db=db)
-        result = query.get_result()
-
-        if result and result.get("rb.branch") == branch.name:
-            await update_relationships_to([result.get("rb_id")], to=delete_at, db=db)
-
-        query = await NodeDeleteQuery.init(db=db, node=self, at=delete_at)
+        query = await NodeDeleteQuery.init(db=db, node=self, at=delete_at, user_id=user_id)
         await query.execute(db=db)
 
         self._node_changelog = node_changelog
@@ -1024,8 +1039,8 @@ async def to_graphql(
                 continue
 
             if field_name == "_updated_at":
-                if self._updated_at:
-                    response[field_name] = await self._updated_at.to_graphql()
+                if updated_at := self._get_updated_at():
+                    response[field_name] = await updated_at.to_graphql()
                 else:
                     response[field_name] = None
                 continue