PostgresInsertCompiler._build_conflict_target()

Complexity

Conditions

6

Size

Total Lines

40

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	6.0106

Importance

Changes	3
Bugs	0	Features	0

1 Method

Rating	Name	Duplication	Size	Complexity
A	PostgresInsertCompiler._assert_valid_field()	0	10	2

from django.core.exceptions import SuspiciousOperation

The import django.core.exceptions could not be resolved.

from django.db.models.sql.compiler import SQLInsertCompiler, SQLUpdateCompiler

The import django.db.models.sql.compiler could not be resolved.

class PostgresReturningUpdateCompiler(SQLUpdateCompiler):

This class has no __init__ method.

    """Compiler for SQL UPDATE statements that return
    the primary keys of the affected rows."""

    def execute_sql(self, _result_type):

The argument _result_type seems to be unused.

        sql, params = self.as_sql()

The Instance of PostgresReturningUpdateCompiler does not seem to have a member named as_sql.

        sql += self._form_returning()

        with self.connection.cursor() as cursor:

The Instance of PostgresReturningUpdateCompiler does not seem to have a member named connection.

            cursor.execute(sql, params)
            primary_keys = cursor.fetchall()

        return primary_keys

    def _form_returning(self):
        """Builds the RETURNING part of the query."""

        qn = self.connection.ops.quote_name

The Instance of PostgresReturningUpdateCompiler does not seem to have a member named connection.

        return ' RETURNING %s' % qn(self.query.model._meta.pk.name)

It seems like _meta was declared protected and should not be accessed from this context.

The Instance of PostgresReturningUpdateCompiler does not seem to have a member named query.

class PostgresInsertCompiler(SQLInsertCompiler):
    """Compiler for SQL INSERT statements."""

    def __init__(self, *args, **kwargs):
        """Initializes a new instance of :see:PostgresInsertCompiler."""

        super().__init__(*args, **kwargs)
        self.qn = self.connection.ops.quote_name

The Instance of PostgresInsertCompiler does not seem to have a member named connection.

    def as_sql(self, return_id=False):
        """Builds the SQL INSERT statement."""

        queries = [
            self._rewrite_insert(sql, params, return_id)
            for sql, params in super().as_sql()
        ]

        return queries

    def execute_sql(self, return_id=False):
        # execute all the generate queries
        with self.connection.cursor() as cursor:

The Instance of PostgresInsertCompiler does not seem to have a member named connection.

            rows = []
            for sql, params in self.as_sql(return_id):
                cursor.execute(sql, params)
                rows.append(cursor.fetchone())

        # create a mapping between column names and column value
        return [
            {
                column.name: row[column_index]
                for column_index, column in enumerate(cursor.description) if row
            }
            for row in rows
        ]

    def _rewrite_insert(self, sql, params, return_id=False):
        """Rewrites a formed SQL INSERT query to include
        the ON CONFLICT clause.

        Arguments:
            sql:
                The SQL INSERT query to rewrite.

            params:
                The parameters passed to the query.

            returning:
                What to put in the `RETURNING` clause
                of the resulting query.

        Returns:
            A tuple of the rewritten SQL query and new params.
        """

        returning = self.qn(self.query.model._meta.pk.name) if return_id else '*'

It seems like _meta was declared protected and should not be accessed from this context.

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        if self.query.conflict_action.value == 'UPDATE':

The Instance of PostgresInsertCompiler does not seem to have a member named query.

            return self._rewrite_insert_update(sql, params, returning)
        elif self.query.conflict_action.value == 'NOTHING':

The Instance of PostgresInsertCompiler does not seem to have a member named query.

            return self._rewrite_insert_nothing(sql, params, returning)

        raise SuspiciousOperation((
            '%s is not a valid conflict action, specify '
            'ConflictAction.UPDATE or ConflictAction.NOTHING.'
        ) % str(self.query.conflict_action))

The Instance of PostgresInsertCompiler does not seem to have a member named query.

    def _rewrite_insert_update(self, sql, params, returning):
        """Rewrites a formed SQL INSERT query to include
        the ON CONFLICT DO UPDATE clause."""

        update_columns = ', '.join([
            '{0} = EXCLUDED.{0}'.format(self.qn(field.column))
            for field in self.query.update_fields

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        ])

        # build the conflict target, the columns to watch
        # for conflicts
        conflict_target = self._build_conflict_target()

        index_predicate = self.query.index_predicate

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        sql_template = (
            '{insert} ON CONFLICT {conflict_target} DO UPDATE '
            'SET {update_columns} RETURNING {returning}'
        )

        if index_predicate:
            sql_template = (
                '{insert} ON CONFLICT {conflict_target} WHERE {index_predicate} DO UPDATE '
                'SET {update_columns} RETURNING {returning}'
            )

        return (
            sql_template.format(
                insert=sql,
                conflict_target=conflict_target,
                update_columns=update_columns,
                returning=returning,
                index_predicate=index_predicate,
            ),
            params
        )

    def _rewrite_insert_nothing(self, sql, params, returning):
        """Rewrites a formed SQL INSERT query to include
        the ON CONFLICT DO NOTHING clause."""

        # build the conflict target, the columns to watch
        # for conflicts
        conflict_target = self._build_conflict_target()

        where_clause = ' AND '.join([
            '{0} = %s'.format(self._format_field_name(field_name))
            for field_name in self.query.conflict_target

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        ])

        where_clause_params = [
            self._format_field_value(field_name)
            for field_name in self.query.conflict_target

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        ]

        params = params + tuple(where_clause_params)

        # this looks complicated, and it is, but it is for a reason... a normal
        # ON CONFLICT DO NOTHING doesn't return anything if the row already exists
        # so we do DO UPDATE instead that never executes to lock the row, and then
        # select from the table in case we're dealing with an existing row..
        return (
            (
                'WITH insdata AS ('
                '{insert} ON CONFLICT {conflict_target} DO UPDATE'
                ' SET id = NULL WHERE FALSE RETURNING {returning})'
                ' SELECT * FROM insdata UNION ALL'
                ' SELECT {returning} FROM {table} WHERE {where_clause} LIMIT 1;'
            ).format(
                insert=sql,
                conflict_target=conflict_target,
                returning=returning,
                table=self.query.objs[0]._meta.db_table,

It seems like _meta was declared protected and should not be accessed from this context.

The Instance of PostgresInsertCompiler does not seem to have a member named query.

                where_clause=where_clause
            ),
            params
        )

    def _build_conflict_target(self):
        """Builds the `conflict_target` for the ON CONFLICT
        clause."""

        conflict_target = []

        if not isinstance(self.query.conflict_target, list):

The Instance of PostgresInsertCompiler does not seem to have a member named query.

            raise SuspiciousOperation((
                '%s is not a valid conflict target, specify '
                'a list of column names, or tuples with column '
                'names and hstore key.'
            ) % str(self.query.conflict_target))

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        def _assert_valid_field(field_name):
            field_name = self._normalize_field_name(field_name)
            if self._get_model_field(field_name):
                return

            raise SuspiciousOperation((
                '%s is not a valid conflict target, specify '
                'a list of column names, or tuples with column '
                'names and hstore key.'
            ) % str(field_name))

        for field_name in self.query.conflict_target:

The Instance of PostgresInsertCompiler does not seem to have a member named query.

            _assert_valid_field(field_name)

            # special handling for hstore keys
            if isinstance(field_name, tuple):
                conflict_target.append(
                    '(%s->\'%s\')' % (
                        self._format_field_name(field_name),
                        field_name[1]
                    )
                )
            else:
                conflict_target.append(
                    self._format_field_name(field_name))

        return '(%s)' % ','.join(conflict_target)

    def _get_model_field(self, name: str):
        """Gets the field on a model with the specified name.

        Arguments:
            name:
                The name of the field to look for.

                This can be both the actual field name, or
                the name of the column, both will work :)

        Returns:
            The field with the specified name or None if
            no such field exists.
        """

        field_name = self._normalize_field_name(name)

        # 'pk' has special meaning and always refers to the primary
        # key of a model, we have to respect this de-facto standard behaviour
        if field_name == 'pk' and self.query.model._meta.pk:

It seems like _meta was declared protected and should not be accessed from this context.

The Instance of PostgresInsertCompiler does not seem to have a member named query.

            return self.query.model._meta.pk

It seems like _meta was declared protected and should not be accessed from this context.

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        for field in self.query.model._meta.local_concrete_fields:

It seems like _meta was declared protected and should not be accessed from this context.

The Instance of PostgresInsertCompiler does not seem to have a member named query.

            if field.name == field_name or field.column == field_name:
                return field

        return None

    def _format_field_name(self, field_name) -> str:
        """Formats a field's name for usage in SQL.

        Arguments:
            field_name:
                The field name to format.

        Returns:
            The specified field name formatted for
            usage in SQL.
        """

        field = self._get_model_field(field_name)
        return self.qn(field.column)

    def _format_field_value(self, field_name) -> str:
        """Formats a field's value for usage in SQL.

        Arguments:
            field_name:
                The name of the field to format
                the value of.

        Returns:
            The field's value formatted for usage
            in SQL.
        """

        field_name = self._normalize_field_name(field_name)
        field = self._get_model_field(field_name)

        return SQLInsertCompiler.prepare_value(
            self,
            field,
            # Note: this deliberately doesn't use `pre_save_val` as we don't
            # want things like auto_now on DateTimeField (etc.) to change the
            # value. We rely on pre_save having already been done by the
            # underlying compiler so that things like FileField have already had
            # the opportunity to save out their data.
            getattr(self.query.objs[0], field.attname)

The Instance of PostgresInsertCompiler does not seem to have a member named query.

        )

    def _normalize_field_name(self, field_name) -> str:

This method could be written as a function/class method.

        """Normalizes a field name into a string by
        extracting the field name if it was specified
        as a reference to a HStore key (as a tuple).

        Arguments:
            field_name:
                The field name to normalize.

        Returns:
            The normalized field name.
        """

        if isinstance(field_name, tuple):
            field_name, _ = field_name

        return field_name