Merge pull request ckan#8684 from JVickery-TBS/feature/datastore-records-return

DataStore Actions Return Records

Author: wardi

changes/8684.feature

@@ -0,0 +1 @@
+`datastore_create`, `datastore_upsert`, and `datastore_delete` now accepts `include_records` to return inserted/deleted records in the return dict, this defaults to False and `datastore_upsert` no longer includes `records` in the return dict by default. `datastore_create`, `datastore_upsert`, and `datastore_delete` with `"include_records": True` will now include the actual data inserted/updated in the return dict's `records` (`deleted_records` for `datastore_delete`). The returned records now include the `_id` column and all field values after processing by any database triggers. Formerly the records returned were only a copy of the values passed by the caller.

ckanext/datastore/backend/postgres.py

@@ -1205,6 +1205,7 @@ def upsert_data(context: Context, data_dict: dict[str, Any]):
     records = data_dict['records']
     sql_columns = ", ".join(
         identifier(name) for name in field_names)
+    return_columns = "_id, " + sql_columns
     if not sql_columns:
         # insert w/ no columns is a postgres error
         return
@@ -1227,16 +1228,21 @@ def upsert_data(context: Context, data_dict: dict[str, Any]):
             rows.append(row)
 
         sql_string = '''INSERT INTO {res_id} ({columns})
-            VALUES ({values});'''.format(
+            VALUES ({values}) {return_statement};'''.format(
             res_id=identifier(data_dict['resource_id']),
             columns=sql_columns,
+            return_statement='RETURNING {return_columns}'.format(
+                return_columns=return_columns) if data_dict[
+                    'include_records'] else '',
             values=', '.join([
                 f":val_{idx}" for idx in range(0, len(field_names))
             ])
         )
 
         try:
-            context['connection'].execute(sa.text(sql_string), rows)
+            results = context['connection'].execute(sa.text(sql_string), rows)
+            if data_dict['include_records']:
+                data_dict['records'] = [dict(r) for r in results.mappings().all()]
         except (DatabaseError, DataError) as err:
             raise ValidationError({
                 'records': [_programming_error_summary(err)],
@@ -1245,7 +1251,7 @@ def upsert_data(context: Context, data_dict: dict[str, Any]):
 
     elif method in [_UPDATE, _UPSERT]:
         unique_keys = _get_unique_key(context, data_dict)
-
+        updated_records = {}
         for num, record in enumerate(records):
             if not unique_keys and '_id' not in record:
                 raise ValidationError({
@@ -1316,12 +1322,16 @@ def upsert_data(context: Context, data_dict: dict[str, Any]):
                 sql_string = u'''
                     UPDATE {res_id}
                     SET ({columns}, "_full_text") = ({values}, NULL)
-                    WHERE ({primary_key}) = ({primary_value});
+                    WHERE ({primary_key}) = ({primary_value})
+                    {return_statement};
                 '''.format(
                     res_id=identifier(data_dict['resource_id']),
                     columns=u', '.join(
                         [identifier(field)
                          for field in used_field_names]),
+                    return_statement='RETURNING {return_columns}'.format(
+                        return_columns=return_columns) if data_dict[
+                            'include_records'] else '',
                     values=u', '.join(values),
                     primary_key=pk_sql,
                     primary_value=pk_values_sql,
@@ -1330,6 +1340,9 @@ def upsert_data(context: Context, data_dict: dict[str, Any]):
                     results = context['connection'].execute(
                         sa.text(sql_string),
                         {**used_values, **unique_values})
+                    if data_dict['include_records']:
+                        for r in results.mappings().all():
+                            updated_records[str(r._id)] = dict(r)
                 except DatabaseError as err:
                     raise ValidationError({
                         'records': [_programming_error_summary(err)],
@@ -1345,42 +1358,43 @@ def upsert_data(context: Context, data_dict: dict[str, Any]):
             elif method == _UPSERT:
                 format_params = dict(
                     res_id=identifier(data_dict['resource_id']),
-                    columns=u', '.join(
+                    columns=('_id, ' if pk_sql == '"_id"' else '') + ', '.join(
                         [identifier(field)
                          for field in used_field_names]),
-                    values=u', '.join([
-                        f'cast(:{p} as nested)'
-                        if field['type'] == 'nested' else ":" + p
-                        for p, field in zip(value_placeholders, used_fields)
-                    ]),
-                    primary_key=pk_sql,
-                    primary_value=pk_values_sql,
+                    set_statement=', '.join(
+                        ['{col}=EXCLUDED.{col}'.format(col=identifier(field))
+                         for field in used_field_names]),
+                    return_statement='RETURNING {return_columns}'.format(
+                        return_columns=return_columns) if data_dict[
+                            'include_records'] else '',
+                    values=('%s, ' % pk_values_sql if pk_sql == '"_id"' else '') +
+                    ', '.join([f'cast(:{p} as nested)' if field['type'] == 'nested'
+                               else ":" + p
+                               for p, field in zip(value_placeholders, used_fields)]),
+                    primary_key=pk_sql
                 )
 
-                update_string = """
-                    UPDATE {res_id}
-                    SET ({columns}, "_full_text") = ({values}, NULL)
-                    WHERE ({primary_key}) = ({primary_value})
-                """.format(**format_params)
-
-                insert_string = """
-                    INSERT INTO {res_id} ({columns})
-                           SELECT {values}
-                           WHERE NOT EXISTS (SELECT 1 FROM {res_id}
-                                    WHERE ({primary_key}) = ({primary_value}))
+                sql_string = """
+                    INSERT INTO {res_id} ({columns}) VALUES ({values})
+                    ON CONFLICT ({primary_key}) DO UPDATE
+                    SET {set_statement}
+                    {return_statement}
                 """.format(**format_params)
 
                 values = {**used_values, **unique_values}
                 try:
-                    context['connection'].execute(
-                        sa.text(update_string), values)
-                    context['connection'].execute(
-                        sa.text(insert_string), values)
+                    results = context['connection'].execute(
+                        sa.text(sql_string), values)
+                    if data_dict['include_records']:
+                        for r in results.mappings().all():
+                            updated_records[str(r._id)] = dict(r)
                 except DatabaseError as err:
                     raise ValidationError({
                         'records': [_programming_error_summary(err)],
                         'records_row': num,
                     })
+        if updated_records:
+            data_dict['records'] = list(updated_records.values())
 
 
 def validate(context: Context, data_dict: dict[str, Any]):
@@ -1410,6 +1424,8 @@ def validate(context: Context, data_dict: dict[str, Any]):
     data_dict_copy.pop('include_total', None)
     data_dict_copy.pop('total_estimation_threshold', None)
     data_dict_copy.pop('records_format', None)
+    data_dict_copy.pop('include_records', None)
+    data_dict_copy.pop('include_deleted_records', None)
     data_dict_copy.pop('calculate_record_count', None)
 
     for key, values in data_dict_copy.items():
@@ -1649,6 +1665,10 @@ def delete_data(context: Context, data_dict: dict[str, Any]):
     validate(context, data_dict)
     fields_types = _get_fields_types(
         context['connection'], data_dict['resource_id'])
+    fields = _get_fields(context['connection'], data_dict['resource_id'])
+    sql_columns = ", ".join(
+        identifier(f['id']) for f in fields)
+    return_columns = "_id, " + sql_columns
 
     query_dict: dict[str, Any] = {
         'where': []
@@ -1659,13 +1679,26 @@ def delete_data(context: Context, data_dict: dict[str, Any]):
                                              fields_types, query_dict)
 
     where_clause, where_values = _where(query_dict['where'])
-    sql_string = u'DELETE FROM "{0}" {1}'.format(
-        data_dict['resource_id'],
-        where_clause
-    )
+    if data_dict['include_deleted_records']:
+        rows_max = config.get('ckan.datastore.search.rows_max')
+        sql_string = '''WITH deleted AS (
+                            DELETE FROM {0} {1} RETURNING {2}
+                        ) SELECT d.* FROM deleted as d LIMIT {3}
+                     '''.format(
+                        identifier(data_dict['resource_id']),
+                        where_clause,
+                        return_columns,
+                        rows_max
+                     )
+    else:
+        sql_string = u'DELETE FROM {0} {1}'.format(
+            identifier(data_dict['resource_id']),
+            where_clause)
 
     try:
-        _execute_single_statement(context, sql_string, where_values)
+        results = _execute_single_statement(context, sql_string, where_values)
+        if data_dict['include_deleted_records']:
+            data_dict['deleted_records'] = [dict(r) for r in results.mappings().all()]
     except ProgrammingError as pe:
         raise ValidationError({'filters': [_programming_error_summary(pe)]})
 

ckanext/datastore/logic/action.py

@@ -62,6 +62,9 @@ def datastore_create(context: Context, data_dict: dict[str, Any]):
     :param records: the data, eg: [{"dob": "2005", "some_stuff": ["a", "b"]}]
                     (optional)
     :type records: list of dictionaries
+    :param include_records: return the full values of inserted records
+                            (optional, default: False)
+    :type include_records: bool
     :param primary_key: fields that represent a unique key (optional)
     :type primary_key: list or comma separated string
     :param indexes: indexes on table (optional)
@@ -197,7 +200,8 @@ def datastore_create(context: Context, data_dict: dict[str, Any]):
 
     result.pop('id', None)
     result.pop('connection_url', None)
-    result.pop('records', None)
+    if not data_dict.pop('include_records', False):
+        result.pop('records', None)
     return result
 
 
@@ -306,6 +310,9 @@ def datastore_upsert(context: Context, data_dict: dict[str, Any]):
     :param records: the data, eg: [{"dob": "2005", "some_stuff": ["a","b"]}]
                     (optional)
     :type records: list of dictionaries
+    :param include_records: return the full values of inserted records
+                            (optional, default: False)
+    :type include_records: bool
     :param method: the method to use to put the data into the datastore.
                    Possible options are: upsert, insert, update
                    (optional, default: upsert)
@@ -359,6 +366,8 @@ def datastore_upsert(context: Context, data_dict: dict[str, Any]):
 
     result.pop('id', None)
     result.pop('connection_url', None)
+    if not data_dict.pop('include_records', False):
+        result.pop('records', None)
 
     if data_dict.get('calculate_record_count', False):
         backend.calculate_record_count(data_dict['resource_id'])  # type: ignore
@@ -460,6 +469,9 @@ def datastore_delete(context: Context, data_dict: dict[str, Any]):
                    If missing delete whole table and all dependent views.
                    (optional)
     :type filters: dictionary
+    :param include_deleted_records: return the full values of deleted records
+                                    (optional, default: False)
+    :type include_deleted_records: bool
     :param calculate_record_count: updates the stored count of records, used to
         optimize datastore_search in combination with the
         `total_estimation_threshold` parameter. If doing a series of requests
@@ -473,7 +485,7 @@ def datastore_delete(context: Context, data_dict: dict[str, Any]):
 
     **Results:**
 
-    :returns: Original filters sent.
+    :returns: Original filters sent and list of deleted_records
     :rtype: dictionary
 
     '''
@@ -537,6 +549,8 @@ def datastore_delete(context: Context, data_dict: dict[str, Any]):
 
     result.pop('id', None)
     result.pop('connection_url', None)
+    if not data_dict.pop('include_deleted_records', False):
+        result.pop('deleted_records', None)
     return result
 
 
@@ -552,6 +566,9 @@ def datastore_records_delete(context: Context, data_dict: dict[str, Any]):
                    If {} delete all records.
                    (required)
     :type filters: dictionary
+    :param include_deleted_records: return the full values of deleted records
+                                    (optional, default: False)
+    :type include_deleted_records: bool
     :param calculate_record_count: updates the stored count of records, used to
         optimize datastore_search in combination with the
         `total_estimation_threshold` parameter. If doing a series of requests

ckanext/datastore/logic/schema.py

@@ -135,6 +135,7 @@ def datastore_create_schema() -> Schema:
                 one_of([u'row'])],
             'function': [not_empty, unicode_only],
         },
+        'include_records': [default(False), boolean_validator],
         'calculate_record_count': [ignore_missing, default(False),
                                    boolean_validator],
         '__junk': [empty],
@@ -150,6 +151,7 @@ def datastore_upsert_schema() -> Schema:
         'id': [ignore_missing],
         'method': [ignore_missing, unicode_safe, one_of(
             ['upsert', 'insert', 'update'])],
+        'include_records': [default(False), boolean_validator],
         'calculate_record_count': [ignore_missing, default(False),
                                    boolean_validator],
         'dry_run': [ignore_missing, boolean_validator],
@@ -164,6 +166,7 @@ def datastore_delete_schema() -> Schema:
         'resource_id': [not_missing, not_empty, unicode_safe],
         'force': [ignore_missing, boolean_validator],
         'id': [ignore_missing],
+        'include_deleted_records': [default(False), boolean_validator],
         'calculate_record_count': [ignore_missing, default(False),
                                    boolean_validator],
         '__junk': [empty],
@@ -178,6 +181,7 @@ def datastore_records_delete_schema() -> Schema:
         'force': [ignore_missing, boolean_validator],
         'filters': [not_missing, dict_only],
         'id': [ignore_missing],
+        'include_deleted_records': [default(False), boolean_validator],
         'calculate_record_count': [ignore_missing, default(False),
                                    boolean_validator],
         '__junk': [empty],

ckanext/datastore/tests/test_create.py

@@ -1457,3 +1457,38 @@ def test_upsert_trigger_exception(self, app):
                 "records": ['"BEANS"? Yeeeeccch!'],
                 "records_row": 1,
             }
+
+    @pytest.mark.ckan_config("ckan.plugins", "datastore")
+    @pytest.mark.usefixtures("clean_datastore", "with_plugins")
+    def test_create_does_not_include_records_by_default(self):
+        package = factories.Dataset()
+        data = {
+            "resource": {"package_id": package["id"]},
+            "fields": [
+                {"id": "movie", "type": "text"},
+                {"id": "director", "type": "text"},
+            ],
+            "force": True,
+            "records": [{"movie": "Cats", "director": "Tom Hooper"}],
+        }
+        result = helpers.call_action("datastore_create", **data)
+        assert 'records' not in result
+
+    @pytest.mark.ckan_config("ckan.plugins", "datastore")
+    @pytest.mark.usefixtures("clean_datastore", "with_plugins")
+    def test_create_include_records_return(self):
+        package = factories.Dataset()
+        data = {
+            "resource": {"package_id": package["id"]},
+            "fields": [
+                {"id": "movie", "type": "text"},
+                {"id": "directors", "type": "text"},
+            ],
+            "force": True,
+            "records": [{"movie": "Cats", "director": "Tom Hooper"}],
+            "include_records": True
+        }
+        result = helpers.call_action("datastore_create", **data)
+        assert 'records' in result
+        for r in result['records']:
+            assert '_id' in r