Organization Validation for Service Inventory

[JVickery-TBS]

We are able to consistently use the releases from the SI github to generate the CSV ref data we need. We have commands which load it into the tables which we define in an sql script in here. The part we were tripped on was relation between Service IDs and Program IDs, but there is no relation that is being defined or maintained currently so don't have to worry about that.

Just extended the temp datastore table to add an optional org_name to it, which we then use in the triggers. The Service ID one works fine, need to make the error message for it still.

The Program IDs one is gunna be interesting as Program IDs is a multiple select.

[wardi]

just those minor things otherwise LGTM

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
302	1	301	0

View the top 1 failed test(s) by shortest run time

ckanext/canada/tests/test_service.py::TestService::test_max_chars

Stack Traces | 0.556s run time

self = <sqlalchemy.engine.base.Connection object at 0x7f767e3d11c0>
dialect = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x7f767f8c5160>
constructor = <bound method DefaultExecutionContext._init_compiled of <class 'sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2'>>
statement = '\n                    INSERT INTO "b163cbed-26c2-4882-ac28-0fcd38b9d8bd" ("fiscal_yr", "service_id", "service_descrip...bd"\n                                    WHERE ("fiscal_yr","service_id") = (%(val_39)s,%(val_40)s))\n                '
parameters = {'val_0': '2022-2023', 'val_1': '1001', 'val_10': 'N', 'val_11': '', ...}
execution_options = immutabledict({'autocommit': symbol('PARSE_AUTOCOMMIT')})
args = (<sqlalchemy.dialects.postgresql.psycopg2.PGCompiler_psycopg2 object at 0x7f767dae2640>, [{'val_0': '2022-2023', 'val_...e_=NullType()), BindParameter('val_12', None, type_=NullType()), BindParameter('val_13', None, type_=NullType()), ...])
kw = {'cache_hit': symbol('CACHE_MISS')}
branched = <sqlalchemy.engine.base.Connection object at 0x7f767e3d11c0>
yp = None
conn = <sqlalchemy.pool.base._ConnectionFairy object at 0x7f767e3f1c10>
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x7f767dae2790>
cursor = <cursor object at 0x7f767dacc4f0; closed: -1>, evt_handled = False

    def _execute_context(
        self,
        dialect,
        constructor,
        statement,
        parameters,
        execution_options,
        *args,
        **kw
    ):
        """Create an :class:`.ExecutionContext` and execute, returning
        a :class:`_engine.CursorResult`."""
    
        branched = self
        if self.__branch_from:
            # if this is a "branched" connection, do everything in terms
            # of the "root" connection, *except* for .close(), which is
            # the only feature that branching provides
            self = self.__branch_from
    
        if execution_options:
            yp = execution_options.get("yield_per", None)
            if yp:
                execution_options = execution_options.union(
                    {"stream_results": True, "max_row_buffer": yp}
                )
    
        try:
            conn = self._dbapi_connection
            if conn is None:
                conn = self._revalidate_connection()
    
            context = constructor(
                dialect, self, conn, execution_options, *args, **kw
            )
        except (exc.PendingRollbackError, exc.ResourceClosedError):
            raise
        except BaseException as e:
            self._handle_dbapi_exception(
                e, util.text_type(statement), parameters, None, None
            )
    
        if (
            self._transaction
            and not self._transaction.is_active
            or (
                self._nested_transaction
                and not self._nested_transaction.is_active
            )
        ):
            self._invalid_transaction()
    
        elif self._trans_context_manager:
            TransactionalContext._trans_ctx_check(self)
    
        if self._is_future and self._transaction is None:
            self._autobegin()
    
        context.pre_exec()
    
        if dialect.use_setinputsizes:
            context._set_input_sizes()
    
        cursor, statement, parameters = (
            context.cursor,
            context.statement,
            context.parameters,
        )
    
        if not context.executemany:
            parameters = parameters[0]
    
        if self._has_events or self.engine._has_events:
            for fn in self.dispatch.before_cursor_execute:
                statement, parameters = fn(
                    self,
                    cursor,
                    statement,
                    parameters,
                    context,
                    context.executemany,
                )
    
        if self._echo:
    
            self._log_info(statement)
    
            stats = context._get_cache_stats()
    
            if not self.engine.hide_parameters:
                self._log_info(
                    "[%s] %r",
                    stats,
                    sql_util._repr_params(
                        parameters, batches=10, ismulti=context.executemany
                    ),
                )
            else:
                self._log_info(
                    "[%s] [SQL parameters hidden due to hide_parameters=True]"
                    % (stats,)
                )
    
        evt_handled = False
        try:
            if context.executemany:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_executemany:
                        if fn(cursor, statement, parameters, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_executemany(
                        cursor, statement, parameters, context
                    )
            elif not parameters and context.no_parameters:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute_no_params:
                        if fn(cursor, statement, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_execute_no_params(
                        cursor, statement, context
                    )
            else:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute:
                        if fn(cursor, statement, parameters, context):
                            evt_handled = True
                            break
                if not evt_handled:
>                   self.dialect.do_execute(
                        cursor, statement, parameters, context
                    )

../................................/venv/lib/python3.9.../sqlalchemy/engine/base.py:1900: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x7f767f8c5160>
cursor = <cursor object at 0x7f767dacc4f0; closed: -1>
statement = '\n                    INSERT INTO "b163cbed-26c2-4882-ac28-0fcd38b9d8bd" ("fiscal_yr", "service_id", "service_descrip...bd"\n                                    WHERE ("fiscal_yr","service_id") = (%(val_39)s,%(val_40)s))\n                '
parameters = {'val_0': '2022-2023', 'val_1': '1001', 'val_10': 'N', 'val_11': '', ...}
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x7f767dae2790>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       psycopg2.errors.RaiseException: TAB-DELIMITED	service_description_fr	This field has a maximum length of {} characters.1800
E       CONTEXT:  PL/pgSQL function service_trigger() line 212 at RAISE

../................................/venv/lib/python3.9.../sqlalchemy/engine/default.py:736: RaiseException

The above exception was the direct cause of the following exception:

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=1316e1c5-5011-48d4-a778-fea8fbca85cf n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f767e3d11c0>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': 'b163cbed-26c2-4882-ac28-0fcd38b9d8bd'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
                    raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
    
                idx_gen = itertools.count()
    
                used_fields = [field for field in fields
                               if field['id'] in record]
    
                used_field_names = _pluck('id', used_fields)
    
                value_placeholders = [
                    f"val_{next(idx_gen)}" for _ in used_field_names
                ]
                values = [":" + p for p in value_placeholders]
                used_values = dict(zip(
                    value_placeholders,
                    [record[field] for field in used_field_names]
                ))
    
                if '_id' in record:
                    placeholder = f'val_{next(idx_gen)}'
                    unique_values = {placeholder: record['_id']}
                    pk_sql = '"_id"'
                    pk_values_sql = ":" + placeholder
                else:
                    placeholders = [
                        f"val_{next(idx_gen)}" for _ in range(len(unique_keys))
                    ]
                    unique_values = dict(zip(
                        placeholders, [record[key] for key in unique_keys]
                    ))
                    pk_sql = ','.join([identifier(part) for part in unique_keys])
                    pk_values_sql = ','.join([":" + p for p in placeholders])
    
                if method == _UPDATE:
                    sql_string = u'''
                        UPDATE {res_id}
                        SET ({columns}, "_full_text") = ({values}, NULL)
                        WHERE ({primary_key}) = ({primary_value});
                    '''.format(
                        res_id=identifier(data_dict['resource_id']),
                        columns=u', '.join(
                            [identifier(field)
                             for field in used_field_names]),
                        values=u', '.join(values),
                        primary_key=pk_sql,
                        primary_value=pk_values_sql,
                    )
                    try:
                        results = context['connection'].execute(
                            sa.text(sql_string),
                            {**used_values, **unique_values})
                    except DatabaseError as err:
                        # (canada fork only): parse constraint sql errors
                        # TODO: upstream contrib!!
                        errmsg = _programming_error_summary(err)
                        if 'violates foreign key constraint' in errmsg:
                            _ = lambda x:x
                            errmsg = _(
                                'Cannot insert records ({refValues}) because '\
                                'they do not exist in the referenced table. '\
                                'Referencing {refKeys} from {refTable}.')
                            raise ValidationError(dict(
                                _parse_constraint_error_from_psql_error(err, errmsg),
                                records_row=num))
                        raise ValidationError({
                            'records': [errmsg],
                            'records_row': num,
                        })
    
                    # validate that exactly one row has been updated
                    if results.rowcount != 1:
                        raise ValidationError({
                            'key': [u'key "{0}" not found'.format(unique_values)]
                        })
    
                elif method == _UPSERT:
                    format_params = dict(
                        res_id=identifier(data_dict['resource_id']),
                        columns=u', '.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,
                    )
    
                    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}))
                    """.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)

...../datastore/backend/postgres.py:1687: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../................................/venv/lib/python3.9.../sqlalchemy/engine/base.py:1380: in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
../................................/venv/lib/python3.9.../sqlalchemy/sql/elements.py:333: in _execute_on_connection
    return connection._execute_clauseelement(
../................................/venv/lib/python3.9.../sqlalchemy/engine/base.py:1572: in _execute_clauseelement
    ret = self._execute_context(
../................................/venv/lib/python3.9.../sqlalchemy/engine/base.py:1943: in _execute_context
    self._handle_dbapi_exception(
../................................/venv/lib/python3.9.../sqlalchemy/engine/base.py:2124: in _handle_dbapi_exception
    util.raise_(
../................................/venv/lib/python3.9.../sqlalchemy/util/compat.py:208: in raise_
    raise exception
../................................/venv/lib/python3.9.../sqlalchemy/engine/base.py:1900: in _execute_context
    self.dialect.do_execute(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x7f767f8c5160>
cursor = <cursor object at 0x7f767dacc4f0; closed: -1>
statement = '\n                    INSERT INTO "b163cbed-26c2-4882-ac28-0fcd38b9d8bd" ("fiscal_yr", "service_id", "service_descrip...bd"\n                                    WHERE ("fiscal_yr","service_id") = (%(val_39)s,%(val_40)s))\n                '
parameters = {'val_0': '2022-2023', 'val_1': '1001', 'val_10': 'N', 'val_11': '', ...}
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x7f767dae2790>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       sqlalchemy.exc.InternalError: (psycopg2.errors.RaiseException) TAB-DELIMITED	service_description_fr	This field has a maximum length of {} characters.1800
E       CONTEXT:  PL/pgSQL function service_trigger() line 212 at RAISE
E       
E       [SQL: 
E                           INSERT INTO "b163cbed-26c2-4882-ac28-0fcd38b9d8bd" ("fiscal_yr", "service_id", "service_description_en", "service_description_fr", "service_type", "service_recipient_type", "service_scope", "client_target_groups", "program_id", "client_feedback_channel", "automated_decision_system", "automated_decision_system_description_en", "automated_decision_system_description_fr", "service_fee", "os_account_registration", "os_authentication", "os_application", "os_decision", "os_issuance", "os_issue_resolution_feedback", "os_comments_client_interaction_en", "os_comments_client_interaction_fr", "last_service_review", "last_service_improvement", "sin_usage", "cra_bn_identifier_usage", "num_phone_enquiries", "num_applications_by_phone", "num_website_visits", "num_applications_online", "num_applications_in_person", "num_applications_by_mail", "num_applications_by_email", "num_applications_by_fax", "num_applications_by_other", "special_remarks_en", "special_remarks_fr", "service_uri_en", "service_uri_fr")
E                                  SELECT %(val_0)s, %(val_1)s, %(val_2)s, %(val_3)s, %(val_4)s, %(val_5)s, %(val_6)s, %(val_7)s, %(val_8)s, %(val_9)s, %(val_10)s, %(val_11)s, %(val_12)s, %(val_13)s, %(val_14)s, %(val_15)s, %(val_16)s, %(val_17)s, %(val_18)s, %(val_19)s, %(val_20)s, %(val_21)s, %(val_22)s, %(val_23)s, %(val_24)s, %(val_25)s, %(val_26)s, %(val_27)s, %(val_28)s, %(val_29)s, %(val_30)s, %(val_31)s, %(val_32)s, %(val_33)s, %(val_34)s, %(val_35)s, %(val_36)s, %(val_37)s, %(val_38)s
E                                  WHERE NOT EXISTS (SELECT 1 FROM "b163cbed-26c2-4882-ac28-0fcd38b9d8bd"
E                                           WHERE ("fiscal_yr","service_id") = (%(val_39)s,%(val_40)s))
E                       ]
E       [parameters: {'val_0': '2022-2023', 'val_1': '1001', 'val_2': 'The Old Age Security (OAS) pension is a monthly payment available to most Canadians 65 years of age who meet the Canadian legal status and residence  ... (147 characters truncated) ... llowance and Allowance for the Survivor. The OAS provides financial support to millions of seniors, including those that are low-income, each year.\n', 'val_3': "Pour les évaluations d'impact, les promoteurs d'un projet sont tenus de présenter à l'AEIC une étude d'impact qui évalue les effets négatifs du proje ... (1520 characters truncated) ... et. La commission d'examen prépare un rapport contenant sa justification, ses conclusions et ses recommandations, et présente ce rapport au ministre.", 'val_4': ['RES'], 'val_5': 'CLIENT', 'val_6': ['EXTERN'], 'val_7': ['PERSON'], 'val_8': ['BGN01'], 'val_9': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], 'val_10': 'N', 'val_11': '', 'val_12': '', 'val_13': 'N', 'val_14': 'Y', 'val_15': 'Y', 'val_16': 'Y', 'val_17': 'Y', 'val_18': 'Y', 'val_19': 'Y', 'val_20': '', 'val_21': '', 'val_22': '', 'val_23': '2021-2022', 'val_24': 'Y', 'val_25': 'Y', 'val_26': 7252346, 'val_27': 0, 'val_28': 5446484, 'val_29': 276390, 'val_30': 0, 'val_31': 792026, 'val_32': 0, 'val_33': 0, 'val_34': 2218002, 'val_35': '- The volume reflected in the \'Applications by Mail\' column include the volume of paper applications for the following OAS pension benefits applica ... (949 characters truncated) ... s reported for CPP and OAS,  such as the volume of calls, are therefore identical and are non-cumulative (i.e. they are not to be added together)."\n', 'val_36': '- Le volume indiqué dans la colonne « Demandes par la poste » comprend le volume de demandes papier pour les types de demandes de prestations de pens ... (1244 characters truncated) ... r le RPC et la SV, comme le volume d\'appels, sont donc identiques et non cumulatives (c\'est-à-dire qu\'elles ne doivent pas être additionnées). "\n', 'val_37': 'https://www.canada..../publicpensions/cpp/old-age-security.html', 'val_38': 'https://www.canada..../pensionspubliques/rpc/securite-vieillesse.html', 'val_39': '2022-2023', 'val_40': '1001'}]
E       (Background on this error at: https://sqlalche..../e/14/2j85)

../................................/venv/lib/python3.9.../sqlalchemy/engine/default.py:736: InternalError

During handling of the above exception, another exception occurred:

up_func = <function datastore_upsert at 0x7f767fd3b160>
context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=1316e1c5-5011-48d4-a778-fea8fbca85cf n...mage_url=None plugin_extras=None>, 'model': <module 'ckan.model' from '.../ckan/model/__init__.py'>, ...}
data_dict = {'resource_id': 'b163cbed-26c2-4882-ac28-0fcd38b9d8bd'}

    @chained_action
    def recombinant_datastore_upsert(up_func: Action,
                                     context: Context,
                                     data_dict: DataDict) -> ChainedAction:
        """
        Wraps datastore_upsert action to split Validation Errors with format_trigger_error.
        """
        try:
>           return up_func(context, data_dict)

...../ckanext/recombinant/logic.py:517: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
...../datastore/logic/action.py:294: in datastore_upsert
    result = backend.upsert(context, data_dict)
...../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
...../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:83: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:70: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=1316e1c5-5011-48d4-a778-fea8fbca85cf n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f767e3d11c0>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': 'b163cbed-26c2-4882-ac28-0fcd38b9d8bd'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
                    raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
    
                idx_gen = itertools.count()
    
                used_fields = [field for field in fields
                               if field['id'] in record]
    
                used_field_names = _pluck('id', used_fields)
    
                value_placeholders = [
                    f"val_{next(idx_gen)}" for _ in used_field_names
                ]
                values = [":" + p for p in value_placeholders]
                used_values = dict(zip(
                    value_placeholders,
                    [record[field] for field in used_field_names]
                ))
    
                if '_id' in record:
                    placeholder = f'val_{next(idx_gen)}'
                    unique_values = {placeholder: record['_id']}
                    pk_sql = '"_id"'
                    pk_values_sql = ":" + placeholder
                else:
                    placeholders = [
                        f"val_{next(idx_gen)}" for _ in range(len(unique_keys))
                    ]
                    unique_values = dict(zip(
                        placeholders, [record[key] for key in unique_keys]
                    ))
                    pk_sql = ','.join([identifier(part) for part in unique_keys])
                    pk_values_sql = ','.join([":" + p for p in placeholders])
    
                if method == _UPDATE:
                    sql_string = u'''
                        UPDATE {res_id}
                        SET ({columns}, "_full_text") = ({values}, NULL)
                        WHERE ({primary_key}) = ({primary_value});
                    '''.format(
                        res_id=identifier(data_dict['resource_id']),
                        columns=u', '.join(
                            [identifier(field)
                             for field in used_field_names]),
                        values=u', '.join(values),
                        primary_key=pk_sql,
                        primary_value=pk_values_sql,
                    )
                    try:
                        results = context['connection'].execute(
                            sa.text(sql_string),
                            {**used_values, **unique_values})
                    except DatabaseError as err:
                        # (canada fork only): parse constraint sql errors
                        # TODO: upstream contrib!!
                        errmsg = _programming_error_summary(err)
                        if 'violates foreign key constraint' in errmsg:
                            _ = lambda x:x
                            errmsg = _(
                                'Cannot insert records ({refValues}) because '\
                                'they do not exist in the referenced table. '\
                                'Referencing {refKeys} from {refTable}.')
                            raise ValidationError(dict(
                                _parse_constraint_error_from_psql_error(err, errmsg),
                                records_row=num))
                        raise ValidationError({
                            'records': [errmsg],
                            'records_row': num,
                        })
    
                    # validate that exactly one row has been updated
                    if results.rowcount != 1:
                        raise ValidationError({
                            'key': [u'key "{0}" not found'.format(unique_values)]
                        })
    
                elif method == _UPSERT:
                    format_params = dict(
                        res_id=identifier(data_dict['resource_id']),
                        columns=u', '.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,
                    )
    
                    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}))
                    """.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)
                    except DatabaseError as err:
                        # (canada fork only): parse constraint sql errors
                        # TODO: upstream contrib!!
                        errmsg = _programming_error_summary(err)
                        if 'violates foreign key constraint' in errmsg:
                            _ = lambda x:x
                            errmsg = _(
                                'Cannot insert records ({refValues}) because '\
                                'they do not exist in the referenced table. '\
                                'Referencing {refKeys} from {refTable}.')
                            raise ValidationError(dict(
                                _parse_constraint_error_from_psql_error(err, errmsg),
                                records_row=num))
>                       raise ValidationError({
                            'records': [errmsg],
                            'records_row': num,
                        })
E                       ckan.logic.ValidationError: None - {'records': [{'service_description_fr': ['This field has a maximum length of 1800 characters.']}], 'records_row': 0}

...../datastore/backend/postgres.py:1702: ValidationError

During handling of the above exception, another exception occurred:

self = <ckanext.canada.tests.test_service.TestService object at 0x7f767f032f40>

    def test_max_chars(self):
        """
        Over max character field values should raise an exception
        """
        chromo = get_chromo('service')
        record = chromo['examples']['record'].copy()
    
        # new line characters \n \r should not be counted
        record['service_description_fr'] = "Pour les évaluations d'impact, les promoteurs d'un projet sont tenus de présenter à l'AEIC une étude d'impact qui évalue les effets négatifs du projet relevant de la compétence fédérale. Après le dépôt de cette étude d'impact par le promoteur, les groupes autochtones et le public sont invités à présenter des commentaires sur le résumé de cette étude d'impact, en particulier sur les effets potentiels et les mesures d'atténuation et de surveillance des effets potentiels négatifs. L'étude d'impact est également examinée par les ministères fédéraux compétents.\r\n\r\nL'AEIC prépare une version provisoire du rapport d'évaluation d'impact contenant sa justifications et ses conclusions sur les effets potentiels du projet, ainsi qu'une version provisoire des conditions. Le public et les groupes autochtones ont la possibilité d'examiner et de commenter ces documents avant qu'ils ne soient transmis au ministre afin de déterminer si les effets négatifs du projet relevant de la compétence fédérale sont dans l'intérêt public.\r\n\r\nLe ministre de l'Environnement et du Changement climatique peut renvoyer une évaluation d'impact à une commission d'examen s'il est dans l'intérêt public de procéder ainsi. Une commission d'examen est un groupe d'experts indépendants qui a pour mandat d'effectuer une évaluation d'impact.\r\n\r\nChaque commission d'examen effectue son analyse de l'étude d'impact présentée par le promoteur. Une commission d'examen doit tenir des audiences publiques pour permettre aux participants de présenter des renseignements, des préoccupations et des commentaires sur les effets potentiels du projet et de poser des questions à propos du projet. La commission d'examen prépare un rapport contenant sa justification, ses conclusions et ses recommandations, et présente ce rapport au ministre."
>       self.lc.action.datastore_upsert(
            resource_id=self.resource_id,
            records=[record])

.../canada/tests/test_service.py:398: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
...../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
...../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
...../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

up_func = <function datastore_upsert at 0x7f767fd3b160>
context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=1316e1c5-5011-48d4-a778-fea8fbca85cf n...mage_url=None plugin_extras=None>, 'model': <module 'ckan.model' from '.../ckan/model/__init__.py'>, ...}
data_dict = {'resource_id': 'b163cbed-26c2-4882-ac28-0fcd38b9d8bd'}

    @chained_action
    def recombinant_datastore_upsert(up_func: Action,
                                     context: Context,
                                     data_dict: DataDict) -> ChainedAction:
        """
        Wraps datastore_upsert action to split Validation Errors with format_trigger_error.
        """
        try:
            return up_func(context, data_dict)
        except ValidationError as e:
            _error_dict = dict(e.error_dict)
            if 'records' not in _error_dict:
                raise
            # type_ignore_reason: incomplete typing
            _error_dict['records'] = list(_error_dict['records'])  # type: ignore
            for record_errs in _error_dict['records']:
                if not isinstance(record_errs, dict):
                    continue
                for field, field_errs in record_errs.items():
                    record_errs[field] = list(format_trigger_error(field_errs))
>           raise ValidationError(_error_dict)
E           ckan.logic.ValidationError: None - {'records': [{'service_description_fr': ['This field has a maximum length of 1800 characters.']}], 'records_row': 0}

...../ckanext/recombinant/logic.py:529: ValidationError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[JVickery-TBS]

@wardi I think this is ready for another review now.

[wardi]

otherwise LGTM