Release updates (#124)

* Session handling for attribute calculation

* Upload records limit fix

* Message for duplicated composite key changed

* adds session handling for attribute calculation thread

---------

Co-authored-by: FelixKirschKern <[email protected]>

Author: lumburovskalina

api/transfer.py

@@ -304,57 +304,75 @@ def __calculate_missing_attributes(project_id: str, user_id: str) -> None:
     notification.send_organization_update(
         project_id=project_id, message="calculate_attribute:started:all"
     )
-    # first check project tokenization completed
-    i = 0
-    while True:
-        i += 1
-        if i >= 60:
-            i = 0
-            ctx_token = general.remove_and_refresh_session(ctx_token, True)
-        if tokenization.is_doc_bin_creation_running(project_id):
-            time.sleep(5)
-            continue
-        else:
-            break
-    # next, ensure that the attributes are calculated and tokenized
-    i = 0
-    while True:
-        time.sleep(1)
-        i += 1
-        if len(attribute_ids) == 0:
-            notification.send_organization_update(
-                project_id=project_id,
-                message="calculate_attribute:finished:all",
-            )
-            break
-        if i >= 60:
-            i = 0
-            ctx_token = general.remove_and_refresh_session(ctx_token, True)
-
-        current_att_id = attribute_ids[0]
-        current_att = attribute.get(project_id, current_att_id)
-        if current_att.state == enums.AttributeState.RUNNING.value:
-            continue
-        elif current_att.state == enums.AttributeState.INITIAL.value:
-            attribute_manager.calculate_user_attribute_all_records(
-                project_id, user_id, current_att_id, True
-            )
-        else:
-            if tokenization.is_doc_bin_creation_running_for_attribute(
-                project_id, current_att.name
-            ):
+
+    try:
+        # first check project tokenization completed
+        i = 0
+        while True:
+            i += 1
+            if i >= 60:
+                i = 0
+                ctx_token = general.remove_and_refresh_session(ctx_token, True)
+            if tokenization.is_doc_bin_creation_running(project_id):
                 time.sleep(5)
                 continue
             else:
-                attribute_ids.pop(0)
-                notification.send_organization_update(
-                    project_id=project_id,
-                    message=f"calculate_attribute:finished:{current_att_id}",
-                )
-        time.sleep(5)
+                break
+        # next, ensure that the attributes are calculated and tokenized
+        i = 0
+        while True:
+            time.sleep(1)
+            i += 1
+            if len(attribute_ids) == 0:
+                break
+            if i >= 60:
+                i = 0
+                ctx_token = general.remove_and_refresh_session(ctx_token, True)
 
-    general.remove_and_refresh_session(ctx_token, False)
-    calculate_missing_embedding_tensors(project_id, user_id)
+            current_att_id = attribute_ids[0]
+            current_att = attribute.get(project_id, current_att_id)
+            if current_att.state == enums.AttributeState.RUNNING.value:
+                continue
+            elif current_att.state == enums.AttributeState.INITIAL.value:
+                attribute_manager.calculate_user_attribute_all_records(
+                    project_id, user_id, current_att_id, True
+                )
+            else:
+                if tokenization.is_doc_bin_creation_running_for_attribute(
+                    project_id, current_att.name
+                ):
+                    time.sleep(5)
+                    continue
+                else:
+                    attribute_ids.pop(0)
+                    notification.send_organization_update(
+                        project_id=project_id,
+                        message=f"calculate_attribute:finished:{current_att_id}",
+                    )
+            time.sleep(5)
+    except Exception as e:
+        print(
+            f"Error while recreating attribute calculation for {project_id} when new records are uploaded : {e}"
+        )
+        get_initial_attributes = attribute.get_all_ordered(
+            project_id,
+            True,
+            state_filter=[
+                enums.AttributeState.INITIAL.value,
+            ],
+        )
+        for attr in get_initial_attributes:
+            attribute.update(
+                project_id, attr.id, state=enums.AttributeState.FAILED.value
+            )
+        general.commit()
+    finally:
+        notification.send_organization_update(
+            project_id=project_id,
+            message="calculate_attribute:finished:all",
+        )
+        general.remove_and_refresh_session(ctx_token, False)
+        calculate_missing_embedding_tensors(project_id, user_id)
 
 
 def calculate_missing_embedding_tensors(project_id: str, user_id: str) -> None:

controller/attribute/manager.py

@@ -1,11 +1,10 @@
-import time
-from typing import List, Optional, Tuple
+from typing import List, Tuple
 from controller.tokenization.tokenization_service import (
     request_tokenize_calculated_attribute,
     request_tokenize_project,
     request_reupload_docbins,
 )
-from submodules.model.business_objects import attribute, record, tokenization
+from submodules.model.business_objects import attribute, record, tokenization, general
 from submodules.model.models import Attribute
 from submodules.model.enums import AttributeState, DataTypes
 from util import daemon, notification
@@ -207,6 +206,7 @@ def calculate_user_attribute_all_records(
 def __calculate_user_attribute_all_records(
     project_id: str, user_id: str, attribute_id: str, include_rats: bool
 ) -> None:
+    session_token = general.get_ctx_token()
     try:
         calculated_attributes = util.run_attribute_calculation_exec_env(
             attribute_id=attribute_id,
@@ -226,6 +226,7 @@ def __calculate_user_attribute_all_records(
             attribute_id=attribute_id,
             log="Attribute calculation failed",
         )
+        general.remove_and_refresh_session(session_token)
         return
 
     util.add_log_to_attribute_logs(
@@ -250,6 +251,7 @@ def __calculate_user_attribute_all_records(
             attribute_id=attribute_id,
             log="Writing to the database failed.",
         )
+        general.remove_and_refresh_session(session_token)
         return
     util.add_log_to_attribute_logs(project_id, attribute_id, "Finished writing.")
 
@@ -262,7 +264,7 @@ def __calculate_user_attribute_all_records(
             request_tokenize_calculated_attribute(
                 project_id, user_id, attribute_item.id, include_rats
             )
-        except:
+        except Exception:
             record.delete_user_created_attribute(
                 project_id=project_id,
                 attribute_id=attribute_id,
@@ -273,6 +275,7 @@ def __calculate_user_attribute_all_records(
                 attribute_id=attribute_id,
                 log="Writing to the database failed.",
             )
+            general.remove_and_refresh_session(session_token)
             return
 
     else:
@@ -292,6 +295,7 @@ def __calculate_user_attribute_all_records(
     notification.send_organization_update(
         project_id, f"calculate_attribute:finished:{attribute_id}"
     )
+    general.remove_and_refresh_session(session_token)
 
 
 def __notify_attribute_calculation_failed(

controller/notification/notification_data.py

@@ -115,7 +115,7 @@
         "docs": enums.DOCS.UPLOADING_DATA.value,
     },
     enums.NotificationType.DUPLICATED_COMPOSITE_KEY.value: {
-        "message_template": "Please upload a file with to the selected composite primary key.",
+        "message_template": "Please upload a file with your projects primary key(s).",
         "title": "Data import",
         "level": enums.Notification.ERROR.value,
         "page": enums.Pages.SETTINGS.value,

controller/transfer/checks.py

@@ -1,4 +1,5 @@
 from typing import Tuple, List, Union, Dict
+from controller.auth import manager as auth_manager
 from controller.misc.config_service import get_config_value
 
 from controller.transfer import util as transfer_util
@@ -128,42 +129,42 @@ def run_checks(df: pd.DataFrame, project_id, user_id) -> None:
 
 
 def run_limit_checks(df: pd.DataFrame, project_id, user_id) -> None:
-    limits = get_config_value("limit_checks")
+    org = auth_manager.get_organization_by_user_id(user_id)
     guard = False
     errors = {}
-    if df.shape[0] > limits["max_rows"]:
+    if df.shape[0] > org.max_rows:
         guard = True
         notification = create_notification(
             NotificationType.NEW_ROWS_EXCEED_MAXIMUM_LIMIT,
             user_id,
             project_id,
             df.shape[0],
-            limits["max_rows"],
+            org.max_rows,
         )
         errors["MaxRows"] = notification.message
     else:
         count_current_records = record.count(project_id)
         if count_current_records:
             updating = get_update_amount(df, project_id)
-            if count_current_records - updating + df.shape[0] > limits["max_rows"]:
+            if count_current_records - updating + df.shape[0] > org.max_rows:
                 guard = True
                 notification = create_notification(
                     NotificationType.TOTAL_ROWS_EXCEED_MAXIMUM_LIMIT,
                     user_id,
                     project_id,
                     count_current_records - updating + df.shape[0],
-                    limits["max_rows"],
+                    org.max_rows,
                 )
                 errors["MaxRows"] = notification.message
 
-    if df.shape[1] > limits["max_cols"]:
+    if df.shape[1] > org.max_cols:
         guard = True
         notification = create_notification(
             NotificationType.COLS_EXCEED_MAXIMUM_LIMIT,
             user_id,
             project_id,
             df.shape[1],
-            limits["max_cols"],
+            org.max_cols,
         )
         errors["MaxCols"] = notification.message
     max_length_dict = dict(
@@ -174,15 +175,15 @@ def run_limit_checks(df: pd.DataFrame, project_id, user_id) -> None:
     )
 
     for key in max_length_dict:
-        if max_length_dict[key] > limits["max_char_count"]:
+        if max_length_dict[key] > org.max_char_count:
             guard = True
             notification = create_notification(
                 NotificationType.COL_EXCEED_MAXIMUM_LIMIT,
                 user_id,
                 project_id,
                 key,
                 max_length_dict[key],
-                limits["max_char_count"],
+                org.max_char_count,
             )
             errors["MaxLength"] = notification.message