backport fix datastore method=insert records_row error value

changes/238.backport.bugfix

@@ -0,0 +1,2 @@
+Fix datastore_upsert method=insert (and datastore_create) records_row error value.
+Used to always report last row regardless of which record contained the error.

changes/9587.bugfix

@@ -0,0 +1,2 @@
+Fix datastore_upsert method=insert (and datastore_create) records_row error value.
+Used to always report last row regardless of which record contained the error.

ckanext/datastore/backend/postgres.py

@@ -1522,8 +1522,30 @@ def upsert_data(context: Context, data_dict: dict[str, Any]):
         )
 
         try:
-            context['connection'].execute(sa.text(sql_string), rows)
+            if len(rows) > 1:
+                itrans = context['connection'].begin_nested()
+                try:
+                    context['connection'].execute(sa.text(sql_string), rows)
+                except:
+                    itrans.rollback()
+                    raise
+                itrans.commit()
+            else:
+                context['connection'].execute(sa.text(sql_string), rows)
+
         except (DatabaseError, DataError) as err:
+            if len(rows) > 1:
+                # determine failing row number for records_row
+                etrans = context['connection'].begin_nested()
+                try:
+                    for num, row in enumerate(rows):
+                        try:
+                            context['connection'].execute(sa.text(sql_string), [row])
+                        except (DatabaseError, DataError) as _err:
+                            break
+                finally:
+                    etrans.rollback()
+
             # (canada fork only): parse constraint sql errors
             # TODO: upstream contrib!!
             errmsg = _programming_error_summary(err)

ckanext/datastore/tests/test_create.py

@@ -1393,6 +1393,7 @@ def test_create_trigger_exception(self, app):
         helpers.call_action(
             u"datastore_function_create",
             name=u"spamexception_trigger",
+            or_replace=True,
             rettype=u"trigger",
             definition=u"""
                 BEGIN
@@ -1408,7 +1409,7 @@ def test_create_trigger_exception(self, app):
                     u"datastore_create",
                     resource={u"package_id": ds["id"]},
                     fields=[{u"id": u"spam", u"type": u"text"}],
-                    records=[{u"spam": u"spam"}, {u"spam": u"EGGS"}],
+                    records=[{"spam": "spam"}, {"spam": "EGGS"}, {"spam": "spam"}],
                     triggers=[{u"function": u"spamexception_trigger"}],
                 )
         assert error.value.error_dict == {u"records": [u'"EGGS"? Yeeeeccch!']}
@@ -1421,6 +1422,7 @@ def test_upsert_trigger_exception(self, app):
         helpers.call_action(
             u"datastore_function_create",
             name=u"spamonly_trigger",
+            or_replace=True,
             rettype=u"trigger",
             definition=u"""
                 BEGIN
@@ -1442,7 +1444,7 @@ def test_upsert_trigger_exception(self, app):
                     u"datastore_upsert",
                     method=u"insert",
                     resource_id=res["resource_id"],
-                    records=[{u"spam": u"spam"}, {u"spam": u"BEANS"}],
+                    records=[{"spam": "spam"}, {"spam": "BEANS"}, {"spam": "spam"}],
                 )
             assert error.value.error_dict == {
                 u"records": [u'"BEANS"? Yeeeeccch!']