ENH: Make get_person customizable in sample upload

sample-management-custom/caosdb-server/scripting/bin/crawl_sample_data_async.py

@@ -47,8 +47,9 @@ from bis_utils import (get_do_not_insert_type_names,
                        return_value_if_not_none,
                        send_mail_with_defaults,
                        SPECIAL_TREATMENT_SAMPLE)
+from sample_helpers.sample_upload_get_person import get_person
 from sample_helpers.sample_upload_post_processing import post_process_samples
-from sample_helpers.utils import get_column_header_name
+from sample_helpers.utils import get_column_header_name, get_entity_name
 
 # suppress warning of diff function
 apilogger = logging.getLogger("linkahead.apiutils")
@@ -396,48 +397,6 @@ def get_nagoya_case(data):
     return nagoya_case
 
 
-def get_person(text) -> db.Record:
-    """
-    Return the BIS ID of the person that is specifed as 'Main User' or 'Sampling Person' in data.
-
-    If the Person is not present in the database, an Exception is raised. Creating new Person Reconrds can only be done by a priviledged user.
-    """
-    # Check in which format the person is identified:
-    person_identifier = text.split(", ")
-    if len(person_identifier) == 1:
-        person = _get_person_by_abbreviation(person_identifier[0])
-    else:
-        person = _get_person_by_fullname(person_identifier[1], person_identifier[0])
-
-    return person
-
-
-def _get_person_by_fullname(first_name, last_name):
-    # seach for person in db
-    res = db.execute_query(
-        "FIND RECORD Person WITH 'First name' = '{}' AND 'Last name' = '{}'".format(first_name, last_name))
-    # if person doesn't exist in database...
-    if len(res) == 0:
-        # There is not enought data in the template to create a new Person record. Hence, we have to raise an Exception
-        error_msg = "There is no person Record with 'First name' = '{}' AND 'Last name' = '{}' in the database. ".format(
-            first_name, last_name)
-        raise DataInconsistencyError(error_msg)
-    else:
-        return res[0]
-
-
-def _get_person_by_abbreviation(abbreviation):
-    # seach for person in db
-    res = db.execute_query("FIND RECORD Person WITH 'Abbreviation' = '{}'".format(abbreviation))
-    # if person doesn't exist in database...
-    if len(res) == 0:
-        # There is not enought data in the template to create a new Person record. Hence, we have to raise an Exception
-        error_msg = "There is no Person Record with Abbreviation = '{}'".format(abbreviation)
-        raise DataInconsistencyError(error_msg)
-    else:
-        return res[0]
-
-
 def synchroize(records, additional_property_ents, htmluserlog_public):
     crawler = Crawler(securityMode=SecurityMode.UPDATE)
     identifiables_definition_file = os.path.expanduser("~/identifiables.yml")

sample-management-custom/caosdb-server/scripting/bin/sample_helpers/sample_upload_get_person.py

+#
+# Copyright (C) 2025 Indiscale GmbH <info@indiscale.com>
+# Copyright (C) 2025 Florian Spreckelsen <f.spreckelsen@indiscale.com>
+#
+# 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/>.
+#
+import linkahead as db
+
+from caosadvancedtools.datainconsistency import DataInconsistencyError
+
+from .utils import get_entity_name
+
+
+def get_person(text: str) -> db.Record:
+    """Return the Person Record that is specifed as 'Main User' or
+    'Sampling Person' in data.
+
+    If the Person is not present in the database, an Exception is
+    raised. Creating new Person Reconrds can only be done by a
+    priviledged user.
+
+    """
+    # Check in which format the person is identified:
+    person_identifier = text.split(", ")
+    if len(person_identifier) == 1:
+        person = _get_person_by_id(person_identifier[0])
+    else:
+        person = _get_person_by_fullname(person_identifier[1], person_identifier[0])
+
+    return person
+
+
+def _get_person_by_fullname(first_name: str, last_name: str) -> db.Record:
+    # seach for person in db
+    res = db.execute_query(
+        f"FIND '{get_entity_name('Person')}' WITH '{
+            get_entity_name('first_name_prop')}'='{first_name}' "
+        f"AND '{get_entity_name('last_name_prop')}'='{last_name}'"
+    )
+    # if person doesn't exist in database...
+    if len(res) == 0:
+        # There is not enought data in the template to create a new Person record. Hence, we have to raise an Exception
+        error_msg = (
+            f"There is no person Record with '{get_entity_name('first_name_prop')}' = '{
+                first_name}' "
+            f"AND '{get_entity_name('last_name_prop')}' = '{last_name}' in the database. ")
+        raise DataInconsistencyError(error_msg)
+    else:
+        return res[0]
+
+
+def _get_person_by_id(eid: str) -> db.Record:
+
+    try:
+        return db.get_entity_by_id(eid, role=get_entity_name("Person"))
+    except db.EmptyUniqueQueryError:
+        raise DataInconsistencyError(
+            f"Could not find a {get_entity_name('Person')} with ID {eid}."
+        )