Re-arrange imports for get_links_of_entities

openatlas/api/resources/model_mapper.py

@@ -2,8 +2,9 @@
 
 from flask import g
 
-from openatlas.api.resources.error import EntityDoesNotExistError, \
-    InvalidCidocClassCodeError, InvalidViewClassError, InvalidSystemClassError
+from openatlas.api.resources.error import (
+    EntityDoesNotExistError, InvalidCidocClassCodeError, InvalidViewClassError,
+    InvalidSystemClassError)
 from openatlas.models.entity import Entity
 from openatlas.models.link import Link
 
@@ -28,12 +29,12 @@ def get_overview_counts() -> dict[str, int]:
     return Entity.get_overview_counts()
 
 
-def get_by_cidoc_classes(class_codes: list[str]) -> list[Entity]:
-    class_codes = list(g.cidoc_classes) \
-        if 'all' in class_codes else class_codes
-    if not all(cc in g.cidoc_classes for cc in class_codes):
+def get_by_cidoc_classes(codes: list[str]) -> list[Entity]:
+    if 'all' in codes:
+        codes = list(g.cidoc_classes)
+    elif not set(codes).issubset(g.cidoc_classes):
         raise InvalidCidocClassCodeError
-    return Entity.get_by_cidoc_class(class_codes, types=True, aliases=True)
+    return Entity.get_by_cidoc_class(codes, types=True, aliases=True)
 
 
 def get_entities_by_view_classes(codes: list[str]) -> list[Entity]:
@@ -57,14 +58,14 @@ def get_all_links_of_entities(
         entities: Union[int, list[int]],
         codes: Union[str, list[str], None] = None) -> list[Link]:
     codes = list(g.properties) if not codes else codes
-    return Link.get_links(entities, codes)
+    return Entity.get_links_of_entities(entities, codes)
 
 
 def get_all_links_of_entities_inverse(
         entities: Union[int, list[int]],
         codes: Optional[Union[str, list[str]]] = None) -> list[Link]:
     codes = list(g.properties) if not codes else codes
-    return Link.get_links(entities, codes, inverse=True)
+    return Entity.get_links_of_entities(entities, codes, inverse=True)
 
 
 def flatten_list_and_remove_duplicates(list_: list[Any]) -> list[Any]:

openatlas/database/entity.py

@@ -417,3 +417,43 @@ def get_linked_entities_recursive(
             """,
             {'id_': id_, 'code': code})
         return [row[first] for row in g.cursor.fetchall()]
+
+    @staticmethod
+    def get_links_of_entities(
+            entities: Union[int, list[int]],
+            codes: Union[str, list[str], None],
+            inverse: bool = False) -> list[dict[str, Any]]:
+        sql = f"""
+            SELECT
+                l.id, l.property_code,
+                l.domain_id,
+                l.range_id,
+                l.description,
+                l.created,
+                l.modified,
+                e.name,
+                l.type_id,
+                COALESCE(to_char(l.begin_from, 'yyyy-mm-dd hh24:mi:ss BC'), '')
+                    AS begin_from, l.begin_comment,
+                COALESCE(to_char(l.begin_to, 'yyyy-mm-dd hh24:mi:ss BC'), '')
+                    AS begin_to,
+                COALESCE(to_char(l.end_from, 'yyyy-mm-dd hh24:mi:ss BC'), '')
+                    AS end_from, l.end_comment,
+                COALESCE(to_char(l.end_to, 'yyyy-mm-dd hh24:mi:ss BC'), '')
+                    AS end_to
+            FROM model.link l
+            JOIN model.entity e
+                ON l.{'domain' if inverse else 'range'}_id = e.id """
+        if codes:
+            codes = codes if isinstance(codes, list) else [codes]
+            sql += ' AND l.property_code IN %(codes)s '
+        sql += f"""
+            WHERE l.{'range' if inverse else 'domain'}_id IN %(entities)s
+            GROUP BY l.id, e.name
+            ORDER BY e.name;"""
+        g.cursor.execute(
+            sql, {
+                'entities': tuple(
+                    entities if isinstance(entities, list) else [entities]),
+                'codes': tuple(codes) if codes else ''})
+        return [dict(row) for row in g.cursor.fetchall()]

openatlas/database/link.py

@@ -62,46 +62,6 @@ def get_linked_entities_inverse(id_: int, codes: list[str]) -> list[int]:
             {'id_': id_, 'codes': tuple(codes)})
         return [row['result_id'] for row in g.cursor.fetchall()]
 
-    @staticmethod
-    def get_links(
-            entities: Union[int, list[int]],
-            codes: Union[str, list[str], None],
-            inverse: bool = False) -> list[dict[str, Any]]:
-        sql = f"""
-            SELECT
-                l.id, l.property_code,
-                l.domain_id,
-                l.range_id,
-                l.description,
-                l.created,
-                l.modified,
-                e.name,
-                l.type_id,
-                COALESCE(to_char(l.begin_from, 'yyyy-mm-dd hh24:mi:ss BC'), '')
-                    AS begin_from, l.begin_comment,
-                COALESCE(to_char(l.begin_to, 'yyyy-mm-dd hh24:mi:ss BC'), '')
-                    AS begin_to,
-                COALESCE(to_char(l.end_from, 'yyyy-mm-dd hh24:mi:ss BC'), '')
-                    AS end_from, l.end_comment,
-                COALESCE(to_char(l.end_to, 'yyyy-mm-dd hh24:mi:ss BC'), '')
-                    AS end_to
-            FROM model.link l
-            JOIN model.entity e
-                ON l.{'domain' if inverse else 'range'}_id = e.id """
-        if codes:
-            codes = codes if isinstance(codes, list) else [codes]
-            sql += ' AND l.property_code IN %(codes)s '
-        sql += f"""
-            WHERE l.{'range' if inverse else 'domain'}_id IN %(entities)s
-            GROUP BY l.id, e.name
-            ORDER BY e.name;"""
-        g.cursor.execute(
-            sql, {
-                'entities': tuple(
-                    entities if isinstance(entities, list) else [entities]),
-                'codes': tuple(codes) if codes else ''})
-        return [dict(row) for row in g.cursor.fetchall()]
-
     @staticmethod
     def delete_by_codes(
             entity_id: int,

openatlas/models/entity.py

@@ -165,7 +165,7 @@ def get_links(
             self,
             codes: Union[str, list[str]],
             inverse: bool = False) -> list[Link]:
-        return Link.get_links(self.id, codes, inverse)
+        return Entity.get_links_of_entities(self.id, codes, inverse)
 
     def delete(self) -> None:
         Entity.delete_(self.id)
@@ -483,3 +483,27 @@ def get_roots(
             ids: list[int],
             inverse: bool = False) -> dict[int, Any]:
         return Db.get_roots(property_code, ids, inverse)
+
+    @staticmethod
+    def get_links_of_entities(
+            entities: Union[int, list[int]],
+            codes: Union[str, list[str], None] = None,
+            inverse: bool = False) -> list[Link]:
+        entity_ids = set()
+        result = Db.get_links_of_entities(
+            entities,
+            codes if isinstance(codes, list) else [str(codes)],
+            inverse)
+        for row in result:
+            entity_ids.add(row['domain_id'])
+            entity_ids.add(row['range_id'])
+        linked_entities = {
+            e.id: e for e in Entity.get_by_ids(entity_ids, types=True)}
+        links = []
+        for row in result:
+            links.append(
+                Link(
+                    row,
+                    domain=linked_entities[row['domain_id']],
+                    range_=linked_entities[row['range_id']]))
+        return links

openatlas/models/link.py

@@ -118,32 +118,6 @@ def get_linked_entity_safe(
             abort(418, f'Missing linked {code} for {id_}')
         return entity
 
-    @staticmethod
-    def get_links(
-            entities: Union[int, list[int]],
-            codes: Union[str, list[str], None] = None,
-            inverse: bool = False) -> list[Link]:
-        from openatlas.models.entity import Entity
-        entity_ids = set()
-        result = Db.get_links(
-            entities,
-            codes if isinstance(codes, list) else [str(codes)],
-            inverse)
-        for row in result:
-            entity_ids.add(row['domain_id'])
-            entity_ids.add(row['range_id'])
-        linked_entities = {
-            entity.id: entity for entity
-            in Entity.get_by_ids(entity_ids, types=True)}
-        links = []
-        for row in result:
-            links.append(
-                Link(
-                    row,
-                    domain=linked_entities[row['domain_id']],
-                    range_=linked_entities[row['range_id']]))
-        return links
-
     @staticmethod
     def delete_by_codes(
             entity: Entity,

openatlas/views/entity.py

@@ -123,12 +123,11 @@ def update(id_: int, copy: Optional[str] = None) -> Union[str, Response]:
     if entity.class_.view in ['artifact', 'place']:
         manager.entity.image_id = manager.entity.get_profile_image_id()
         if not manager.entity.image_id:
-            for link_ in manager.entity.get_links('P67', inverse=True):
-                domain = link_.domain
-                if domain.class_.view == 'file' \
-                        and get_base_table_data(domain)[3] \
+            for l in manager.entity.get_links('P67', inverse=True):
+                if l.domain.class_.view == 'file' \
+                        and get_base_table_data(l.domain)[3] \
                         in app.config['DISPLAY_FILE_EXTENSIONS']:
-                    manager.entity.image_id = domain.id
+                    manager.entity.image_id = l.domain.id
                     break
     return render_template(
         'entity/update.html',