_generate_triggers()

Complexity

Conditions

16

Size

Total Lines

114

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	255.6934

Importance

Changes	4
Bugs	0	Features	0

2 Methods

Rating	Name	Duplication	Size	Complexity
D	_analyse_clause()	0	68	10
F	_query_compiling()	0	105	15

# ~*~ coding: utf-8 ~*~
#-
# OSMAlchemy - OpenStreetMap to SQLAlchemy bridge
# Copyright (c) 2016 Dominik George <nik@naturalnet.de>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Alternatively, you are free to use OSMAlchemy under Simplified BSD, The
# MirOS Licence, GPL-2+, LGPL-2.1+, AGPL-3+ or the same terms as Python
# itself.

""" Trigger code for live OSMAlchemy/Overpass integration. """

import datetime

The import datetime seems to be unused.

import operator
from sqlalchemy import inspect

Unused inspect imported from sqlalchemy

from sqlalchemy.sql.elements import BinaryExpression, BooleanClauseList, BindParameter
from sqlalchemy.event import listens_for
from sqlalchemy.orm import Query
from sqlalchemy.sql.annotation import AnnotatedColumn

The name AnnotatedColumn does not seem to exist in module sqlalchemy.sql.annotation.

from weakref import WeakSet

standard import "from weakref import WeakSet" comes before "from sqlalchemy import inspect"

from .online import _get_single_element_by_id

Unused _get_single_element_by_id imported from online

from .util import _import_osm_xml

Unused _import_osm_xml imported from util

def _generate_triggers(osmalchemy, maxage=60*60*24):

The argument maxage seems to be unused.

    """ Generates the triggers for online functionality.

      osmalchemy - reference to the OSMAlchemy instance to be configured
      maxage - maximum age of objects before they are updated online, in seconds
    """

    _visited_queries = WeakSet()

    @listens_for(Query, "before_compile")
    def _query_compiling(query):

The variable _query_compiling seems to be unused.

        # Get the session associated with the query:
        session = query.session

The variable session seems to be unused.

        # Prevent recursion by skipping already-seen queries
        if query in _visited_queries:
            return
        else:
            _visited_queries.add(query)

        # Check whether this query affects our model
        affected_models = set([c["type"] for c in query.column_descriptions])
        our_models = set([osmalchemy.Node, osmalchemy.Way,  osmalchemy.Relation,

Exactly one space required after comma

                          osmalchemy.Element])
        if affected_models.isdisjoint(our_models):
            # None of our models is affected
            return

        # Check whether this query filters elements
        # Online update will only run on a specified set, not all data
        if query.whereclause is None:
            # No filters
            return

        # Define operator to string mapping
        _ops = {operator.eq: "==",
                operator.ne: "!=",
                operator.lt: "<",
                operator.gt: ">",
                operator.le: "<=",
                operator.ge: ">=",
                operator.and_: "&&",
                operator.or_: "||"}

        # Traverse whereclause recursively
        def _analyse_clause(clause):

Too many return statements (8/6)

            if type(clause) is BinaryExpression:

Using type() instead of isinstance() for a typecheck.

                # This is something like "latitude >= 51.0"
                left = clause.left
                right = clause.right
                op = clause.operator

The name op does not conform to the variable naming conventions ([a-z_][a-z0-9_]{2,30}$).

                # Left part should be a column
                if type(left) is AnnotatedColumn:

Using type() instead of isinstance() for a typecheck.

                    # Get table class and field
                    model = left._annotations["parentmapper"].class_

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

                    field = left

                    # Double-check this model belongs to us
                    if model in our_models:
                        # Convert model class and field to string names
                        left = (model.__name__, field.name)
                    else:
                        return None
                else:
                    # Right now, we cannot cope with anything but a column on the left
                    return None

                # Right part should be a literal value
                if type(right) is BindParameter:

Using type() instead of isinstance() for a typecheck.

                    # Extract literal value
                    right = right.value
                else:
                    # Right now, we cannot cope with something else here
                    return None

                # Look for a known operator
                if op in _ops.keys():
                    # Get string representation
                    op = _ops[op]

The name op does not conform to the variable naming conventions ([a-z_][a-z0-9_]{2,30}$).

                else:
                    # Right now, we cannot cope with other operators
                    return None

                # Return polish notation tuple of this clause
                return (op, left, right)
            elif type(clause) is BooleanClauseList:

Using type() instead of isinstance() for a typecheck.

                # This is an AND or OR operation
                op = clause.operator

The name op does not conform to the variable naming conventions ([a-z_][a-z0-9_]{2,30}$).

                clauses = []

                # Iterate over all the clauses in this operation
                for clause in clause.clauses:
                    # Recursively analyse clauses
                    res = _analyse_clause(clause)
                    # None is returned for unsupported clauses or operations
                    if res is not None:
                        # Append polish notation result to clauses list
                        clauses.append(res)

                # Look for a known operator
                if op in _ops.keys():
                    # Get string representation
                    op = _ops[op]

The name op does not conform to the variable naming conventions ([a-z_][a-z0-9_]{2,30}$).

                else:
                    # Right now, we cannot cope with anything else
                    return None

                # Return polish notation tuple of this clause
                return (op, clauses)
            else:
                # We hit an unsupported type of clause
                return None
        tree = _analyse_clause(query.whereclause)

The variable tree seems to be unused.