Changeset 3d598d5 in flexoentity

Timestamp:

12/08/25 15:35:46 (5 weeks ago)

Author:

Enrico Schwass <ennoausberlin@…>

Branches:

master

Children:

ef5aac4

Parents:

e230c40

Message:

promotion methods added

Location:

flexoentity

Files:

• domain_manager.py (modified) (1 diff)
• entity_manager.py (modified) (3 diffs)

flexoentity/domain_manager.py

@@ -1 +1 @@
 from __future__ import annotations
 import json
-
-from typing import Iterable
 
 from .flexo_collection import FlexoCollection

flexoentity/entity_manager.py

@@ -1 +1 @@
 from .flexo_entity import FlexoEntity
 from .typed_collection import TypedCollection
+
+
+class EntityNotFoundError(KeyError):
+    pass
+
+
+class InvalidEntityError(ValueError):
+    pass
+
+
+class DuplicateEntityError(ValueError):
+    pass
 
 
@@ -26 +38 @@
         self.staging_backend = staging_backend
         self.permanent_backend = permanent_backend
+
+    def backend_of_domain_id(self, domain_id: str):
+        for backend_name, backend in [
+                ("local", self.local_backend),
+                ("staging", self.staging_backend),
+                ("permanent", self.permanent_backend),
+        ]:
+            for dom in backend.load_all():
+                if dom.domain_id == domain_id:
+                    return backend_name
+        return None
+
+    def backend_of_flexo_id(self, flexo_id: str):
+        for backend_name, backend in [
+                ("local", self.local_backend),
+                ("staging", self.staging_backend),
+                ("permanent", self.permanent_backend),
+        ]:
+            if backend.load(flexo_id) is not None:
+                return backend_name
+        return None
 
     # ------------------------------------------------------------------
@@ -58 +91 @@
     def all(self):
         """Load all entities as a list of instances."""
-        return self.local_backend.load_all()
+        return (self.local_backend.load_all() + self.staging_backend.load_all() +
+                self.permanent_backend.load_all())
 
     def promote_to_staging(self, flexo_id):
         entity = self.local_backend.load(flexo_id)
+        if entity is None:
+            raise EntityNotFoundError
         self.staging_backend.save(entity)
+        self.local_backend.delete(flexo_id)
 
     def promote_to_permanent(self, flexo_id):
         entity = self.staging_backend.load(flexo_id)
+        if entity is None:
+            raise EntityNotFoundError
         self.permanent_backend.save(entity)
+        self.staging_backend.delete(flexo_id)
 
     def sync_all(self):