PostgresInsertCompiler._assert_valid_field() - Code Metrics - Inspection of "Make ConditionalUniqueIndexes actually usable." - SectorLabs/django-postgres-extra - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#29)

by Alexandru

created 2017-11-13 10:38 UTC

PostgresInsertCompiler._assert_valid_field() A

↳ Parent: PostgresInsertCompiler._build_conflict_target()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	2
c	3
b	0
f	0
dl	0
loc	10
rs	9.4285
ccs	5
cts	5
cp	1
crap	2

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} WHERE {index_predicate} DO UPDATE'
                ' SET {update_columns} RETURNING {returning}'
            ) if index_predicate else (
                '{insert} ON CONFLICT {conflict_target} 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,
            getattr(self.query.objs[0], field_name)

        )

    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


Pull Request — master (#29)

PostgresInsertCompiler._assert_valid_field() A

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing init.py files

1. Missing Dependencies

2. Missing init.py files

1	1	from django.core.exceptions import SuspiciousOperation
		0 ignored issues – show Configuration introduced 2017-02-24 13:59 UTC by Report Bug Copy Issue Report The import `django.core.exceptions` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
2	1	from django.db.models.sql.compiler import SQLInsertCompiler, SQLUpdateCompiler
		0 ignored issues – show Configuration introduced 2017-02-23 14:02 UTC by Report Bug Copy Issue Report The import `django.db.models.sql.compiler` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
3
4
5	1	class PostgresReturningUpdateCompiler(SQLUpdateCompiler):
		0 ignored issues – show Coding Style introduced 2017-04-08 11:58 UTC by Report Bug Copy Issue Report This class has no `__init__` method. Loading history...
6		"""Compiler for SQL UPDATE statements that return
7		the primary keys of the affected rows."""
8
9	1	def execute_sql(self, _result_type):
		0 ignored issues – show Unused Code introduced 2017-03-05 13:41 UTC by Report Bug Copy Issue Report The argument `_result_type` seems to be unused. Loading history...
10	1	sql, params = self.as_sql()
		0 ignored issues – show Bug introduced 2017-04-08 11:58 UTC by Report Bug Copy Issue Report The Instance of `PostgresReturningUpdateCompiler` does not seem to have a member named `as_sql`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
11	1	sql += self._form_returning()
12
13	1	with self.connection.cursor() as cursor:
		0 ignored issues – show Bug introduced 2017-04-08 11:58 UTC by Report Bug Copy Issue Report The Instance of `PostgresReturningUpdateCompiler` does not seem to have a member named `connection`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
14	1	cursor.execute(sql, params)
15	1	primary_keys = cursor.fetchall()
16
17	1	return primary_keys
18
19	1	def _form_returning(self):
20		"""Builds the RETURNING part of the query."""
21
22	1	qn = self.connection.ops.quote_name
		0 ignored issues – show Bug introduced 2017-04-08 11:58 UTC by Report Bug Copy Issue Report The Instance of `PostgresReturningUpdateCompiler` does not seem to have a member named `connection`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
23	1	return ' RETURNING %s' % qn(self.query.model._meta.pk.name)
		0 ignored issues – show Coding Style Best Practice introduced 2017-03-05 13:41 UTC by Report Bug Copy Issue Report It seems like `_meta` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: 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 Loading history... Bug introduced 2017-04-08 11:58 UTC by Report Bug Copy Issue Report The Instance of `PostgresReturningUpdateCompiler` does not seem to have a member named `query`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
24
25
26	1	class PostgresInsertCompiler(SQLInsertCompiler):
27		"""Compiler for SQL INSERT statements."""
28
29	1	def __init__(self, args, *kwargs):
30		"""Initializes a new instance of :see:PostgresInsertCompiler."""
31
32	1	super().__init__(args, *kwargs)

SectorLabs / django-postgres-extra

Pull Request — master (#29)

PostgresInsertCompiler._assert_valid_field() A

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