PostgresInsertCompiler._format_field_value() - Code Metrics - Inspection of "Merge pull request #46 from PeterJCLaw/fix-foreign..." - SectorLabs/django-postgres-extra - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6eb2f8...0124f0 )

by Swen

created 2018-01-29 09:28 UTC

PostgresInsertCompiler._format_field_value() B

↳ Parent: PostgresInsertCompiler

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	25
rs	8.8571
ccs	4
cts	4
cp	1
crap	1

from django.core.exceptions import SuspiciousOperation
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
from django.db.models.sql.compiler import SQLInsertCompiler, SQLUpdateCompiler
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3


class PostgresReturningUpdateCompiler(SQLUpdateCompiler):

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

    def execute_sql(self, _result_type):

        sql, params = self.as_sql()

        sql += self._form_returning()

        with self.connection.cursor() as cursor:

            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

        return ' RETURNING %s' % qn(self.query.model._meta.pk.name)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent


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


    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:

            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 '*'
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent

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

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

            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))


    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

        ])

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

        index_predicate = self.query.index_predicate


        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

        ])

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

        ]

        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,
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                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):

            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))


        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:

            _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:
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
            return self.query.model._meta.pk
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent

        for field in self.query.model._meta.local_concrete_fields:
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
            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)

        )

    def _normalize_field_name(self, field_name) -> str:
class Foo:
    def some_method(self, x, y):
        return x + y;
        """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


Push — master ( 6eb2f8...0124f0 )

PostgresInsertCompiler._format_field_value() B

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing init.py files

1. Missing Dependencies

2. Missing init.py files

SectorLabs / django-postgres-extra

Push — master ( 6eb2f8...0124f0 )

PostgresInsertCompiler._format_field_value() B

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files

2. Missing init.py files