_import_osm_dom()

Complexity

Conditions

32

Size

Total Lines

136

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	79
CRAP Score	32.0154

Importance

Changes	4
Bugs	0	Features	0

5 Methods

Rating	Name	Duplication	Size	Complexity
C	_dom_attrs_to_any()	0	13	8
B	_dom_to_way()	0	25	4
F	_dom_to_relation()	0	42	9
A	_dom_tags_to_any()	0	9	3
A	_dom_to_node()	0	21	3

# ~*~ 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.

""" Utility code for the OSMAlchemy database OSMAlchemy. """

import dateutil.parser
import operator

standard import "import operator" comes before "import dateutil.parser"

The import operator seems to be unused.

import xml.dom.minidom as minidom

standard import "import xml.dom.minidom as minidom" comes before "import dateutil.parser"

from sqlalchemy.sql.elements import BinaryExpression, BooleanClauseList, BindParameter

Unused BinaryExpression imported from sqlalchemy.sql.elements

Unused BooleanClauseList imported from sqlalchemy.sql.elements

Unused BindParameter imported from sqlalchemy.sql.elements

from sqlalchemy.sql.annotation import AnnotatedColumn

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

Unused AnnotatedColumn imported from sqlalchemy.sql.annotation

def _import_osm_dom(osma, session, dom):
    """ Import a DOM tree from OSM XML into an OSMAlchemy model.

    Not called directly; used by _import_osm_xml and _import_osm_file.
    """

    def _dom_attrs_to_any(e, element):

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

        if "version" in e.attributes.keys():
            element.version = int(e.attributes["version"].value)
        if "changeset" in e.attributes.keys():
            element.changeset = int(e.attributes["changeset"].value)
        if "user" in e.attributes.keys():
            element.user = e.attributes["user"].value
        if "uid" in e.attributes.keys():
            element.uid = int(e.attributes["uid"].value)
        if "visible" in e.attributes.keys():
            element.visible = True if e.attributes["visible"].value == "true" else False
        if "timestamp" in e.attributes.keys():
            element.timestamp = dateutil.parser.parse(e.attributes["timestamp"].value)

    def _dom_tags_to_any(e, element):

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

        # Remove all tags previously associated with element
        for k in list(element.tags.keys()):
            del element.tags[k]

        # Iterate over all <tag /> nodes in the DOM element
        for t in e.getElementsByTagName("tag"):

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

            # Append data to tags
            element.tags[t.attributes["k"].value] = t.attributes["v"].value

    def _dom_to_node(e):

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

        with session.no_autoflush:
            # Get mandatory node id
            id = int(e.attributes["id"].value)

This seems to re-define the built-in id.

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

            # Find object in database and create if non-existent
            node = session.query(osma.node).filter_by(id=id).scalar()
            if node is None:
                node = osma.node(id=id)

            # Store mandatory latitude and longitude
            node.latitude = e.attributes["lat"].value
            node.longitude = e.attributes["lon"].value

            # Store other attributes and tags
            _dom_attrs_to_any(e, node)
            _dom_tags_to_any(e, node)

        # Add to session
        session.add(node)
        session.commit()

    def _dom_to_way(e):

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

        with session.no_autoflush:
            # Get mandatory way id
            id = int(e.attributes["id"].value)

This seems to re-define the built-in id.

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

            # Find object in database and create if non-existent
            way = session.query(osma.way).filter_by(id=id).scalar()
            if way is None:
                way = osma.way(id=id)

            # Find all related nodes
            for n in e.getElementsByTagName("nd"):

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

                # Get node id and find object
                ref = int(n.attributes["ref"].value)
                node = session.query(osma.node).filter_by(id=ref).one()
                # Append to nodes in way
                way.nodes.append(node)

            # Store other attributes and tags
            _dom_attrs_to_any(e, way)
            _dom_tags_to_any(e, way)

        # Add to session
        session.add(way)
        session.commit()

    def _dom_to_relation(e):

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

        with session.no_autoflush:
            # Get mandatory way id
            id = int(e.attributes["id"].value)

This seems to re-define the built-in id.

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

            # Find object in database and create if non-existent
            relation = session.query(osma.relation).filter_by(id=id).scalar()
            if relation is None:
                relation = osma.relation(id=id)

            # Find all members
            for m in e.getElementsByTagName("member"):

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

                # Get member attributes
                ref = int(m.attributes["ref"].value)
                type = m.attributes["type"].value

This seems to re-define the built-in type.

                if "role" in m.attributes.keys():
                    role = m.attributes["role"].value
                else:
                    role = ""
                element = session.query(osma.element).filter_by(id=ref, type=type).scalar()
                if element is None:
                    # We do not know the member yet, create a stub
                    if type == "node":
                        element = osma.node(id=ref)
                    elif type == "way":
                        element = osma.way(id=ref)
                    elif type == "relation":
                        element = osma.relation(id=ref)
                    # We need to commit here because element could be repeated
                    session.add(element)
                    session.commit()
                # Append to members
                relation.members.append((element, role))

            # Store other attributes and tags
            _dom_attrs_to_any(e, relation)
            _dom_tags_to_any(e, relation)

        # Add to session
        session.add(relation)
        session.commit()

    # Get root element
    osm = dom.documentElement

    # Iterate over children to find nodes, ways and relations
    for e in osm.childNodes:

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

        # Determine element type
        if e.nodeName == "node":
            _dom_to_node(e)
        elif e.nodeName == "way":
            _dom_to_way(e)
        elif e.nodeName == "relation":
            _dom_to_relation(e)

        # Rmove children
        e.unlink()

def _import_osm_xml(osma, session, xml):
    """ Import a string in OSM XML format into an OSMAlchemy model.

      osma - reference to the OSMAlchemy model instance
      session - an SQLAlchemy session
      xml - string containing the XML data
    """

    # Parse string into DOM structure
    dom = minidom.parseString(xml)

    return _import_osm_dom(osma, session, dom)

def _import_osm_file(osma, session, file):
    """ Import a file in OSM XML format into an OSMAlchemy model.

      osma - reference to the OSMAlchemy model instance
      session - an SQLAlchemy session
      path - path to the file to import or open file object
    """

    # Parse document into DOM structure
    dom = minidom.parse(file)

    return _import_osm_dom(osma, session, dom)