up

src/caoscrawler/sync_node.py

+#!/usr/bin/env python3
+# encoding: utf-8
+#
+# This file is a part of the LinkAhead Project.
+#
+# Copyright (C) 2024 Henrik tom Wörden
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+#
+
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Union
+from uuid import uuid4 as uuid
+
+import linkahead as db
+
+
+class SyncNode():
+    """ represents the information related to an Entity as it shall be created in LinkAhead
+
+    The following information is taken from db.Entity object during initialization or when the
+    object is updated using `update(entity)`:
+    - id
+    - role
+    - parents
+    - path
+    - name
+    - description
+    - properties
+
+    Typically, this class is used in the following way:
+    1. A SyncNode is initialized with a db.Entity object
+    2. The SyncNode object is possibly updated one or more times with further db.Entity objects
+    3. A db.Entity object is created (`export_entity`) that contains the combined information of
+       the previous db.Entity objects.
+    """
+
+    def __init__(self, entity: db.Entity, registered_identifiable: Optional[db.RecordType] =
+                 None) -> None:
+        self.id = entity.id
+        self.role = entity.role
+        self.parents = entity.parents
+        self.path = entity.path
+        self.name = entity.name
+        self.description = entity.description
+        self.properties = list(entity.properties)
+        self.uuid = uuid()
+        self.identifiable = None
+        self.registered_identifiable = registered_identifiable
+
+    def update(self, other: SyncNode) -> None:
+        if other.identifiable is not None and self.identifiable is not None:
+            assert (other.identifiable.get_representation() ==
+                    self.identifiable.get_representation())
+        for attr in ["id", "path", "role", "path", "name", "description"]:
+            if other.__getattribute__(attr) is not None:
+                if self.__getattribute__(attr) is None:
+                    self.__setattr__(attr, other.__getattribute__(attr))
+                else:
+                    assert self.__getattribute__(attr) == other.__getattribute__(attr)
+        for p in other.parents:
+            if p not in self.parents:
+                self.parents.append(p)
+        for p in other.properties:
+            if p not in self.properties:
+                self.properties.append(p)
+
+    def export_entity(self) -> db.Entity:
+        ent = None
+        if self.role == "Record":
+            ent = db.Record()
+        elif self.role == "File":
+            ent = db.File()
+        else:
+            raise RuntimeError("Invalid role")
+        for attr in ["id", "path", "role", "path", "name", "description"]:
+            ent.__setattr__(attr, self.__getattribute__(attr))
+        for p in self.parents:
+            ent.add_parent(p)
+        for p in self.properties:
+            if ent.get_property(p) is not None:
+                if ent.get_property(p).value != p.value:
+                    raise Exception()
+            else:
+                ent.add_property(p)
+        return ent