Python.LocalDB.query.check_value() - Code Metrics - Inspection of "Dicom Anonymization" - CNBP/DICOMTransit - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#2)

by Yang

created 2018-08-01 16:22 UTC

Python.LocalDB.query.check_value() A

↳ Parent: Python.LocalDB.query

Complexity

Conditions

Size

Total Lines	42
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	42
rs	9.5
c	0
b	0
f	0
cc	4
nop	4

import sys
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
import logging
import sqlite3
from pathlib import Path
from LocalDB.schema import CNBP_schema


logging.basicConfig(stream=sys.stdout, level=logging.INFO)


def check_value(database_path, table_name, ColumnName, ColumnValue):

    """
    Check if a subject exist in the given database and given table
    :param database_path: path to the SQLite database
    :param table_name: the name of the table being queried
    :param ColumnName: the column being queried
    :param ColumnValue: the value of the column being checked
    :return: boolean on if this is ever found in the given database in the given table, in the given column.

    """
    logger = logging.getLogger('LORISQuery_CheckSubjectExist')


    SQLPath = Path(database_path)


    # check if path is a file and exist.
    if not SQLPath.is_file():
        logger.info('SQLite database file does not exist!')
        return False

    # Try to connect the database to start the process:
    try:
        # Create on Connecting to the database file
        ConnectedDatabase = sqlite3.connect(database_path)

        c = ConnectedDatabase.cursor()


        logger.info("Checking key value: " + str(ColumnValue) + " in " + ColumnName + " in SQLite database.")


        # Creating a new SQLite table_name with DBKey column (inspired by: https://sebastianraschka.com/Articles/2014_sqlite_in_python_tutorial.html)

        c.execute('SELECT * FROM {table_name} WHERE {columnname}="{columnvalue}"'.format(table_name=table_name, columnname=ColumnName, columnvalue=ColumnValue))


        result_rows = c.fetchall()

    except Exception as e:

        raise IOError()

    # Closing the connection to the database file
    ConnectedDatabase.close()

    if len(result_rows) > 0:

        return True, result_rows
    else:
        return False, result_rows


def create_entry(database_path, table_name, key_field, key_field_value):
    """
    A general function to create entries into the database BY providing the name of the KEYValue field and KEYvalue value to be created

    Note it MUST be the keyfield.
    :param database_path: path to the database
    :param table_name: name of the table
    :param key_field: KeyFiled in the table to be created
    :param key_field_value: value of the key_field to be created. 

    :return: if the entry has been successfully created.
    """
    logger = logging.getLogger('LORISQuery_CreateSubject')

    # if SQL already exist, quit script.
    SQLPath = Path(database_path)


    # check if path is a file and exist.
    if not SQLPath.is_file():
        logger.info('SQLite database file does not exist!')
        return False

    # Try to connect the database to start the process:
    try:
        # Create on Connecting to the database file
        ConnectedDatabase = sqlite3.connect(database_path)

        c = ConnectedDatabase.cursor()


        logger.info('Creating new record in SQLite database.')

        # Creating a new SQLite record row (inspired by: https://sebastianraschka.com/Articles/2014_sqlite_in_python_tutorial.html)

        c.execute('INSERT OR IGNORE INTO {tn} ({field}) VALUES ("{value}")'.format(tn=table_name, field=key_field, value=key_field_value))

    except Exception as e:

        raise IOError()

    # Closing the connection to the database file
    ConnectedDatabase.commit()
    ConnectedDatabase.close()
    return True


def update_entry(database_path, table_name, key_field, key_field_value, field, field_value):

    """
    A general function to create entries into the database BY providing the name of the KEYValue field and KEYvalue value to be created

    :param database_path:
    :param table_name:
    :param key_field:
    :param key_field_value:
    :param field:
    :param field_value:
    :return:
    """

    logger = logging.getLogger('LORISQuery_CreateSubject')

    # if SQL already exist, quit script.
    SQLPath = Path(database_path)


    # check if path is a file and exist.
    if not SQLPath.is_file():
        logger.info('SQLite database file does not exist!')
        return False

    # Try to connect the database to start the process:
    try:
        # Create on Connecting to the database file
        ConnectedDatabase = sqlite3.connect(database_path)

        c = ConnectedDatabase.cursor()


        logger.info('Update records in SQLite database.')

        # Update SQLite record row where key field values are found (inspired by: https://sebastianraschka.com/Articles/2014_sqlite_in_python_tutorial.html)

        c.execute('UPDATE {tn} SET {f}="{fv}" WHERE {kf}="{kfv}"'.format(tn=table_name, f=field, fv=field_value, kf=key_field, kfv=key_field_value))


    except Exception as e:

        raise IOError()

    # Closing the connection to the database file
    ConnectedDatabase.commit()
    ConnectedDatabase.close()


def check_header(database_path, table_name):
    """
    Finds the table, connect to it, and then return the header.
    :param database_path:
    :param table_name:
    :return:
    """

    logger = logging.getLogger('SQLite check_header check')


    table_header = None

    try:
        # Create on Connecting to the database file
        ConnectedDatabase = sqlite3.connect(database_path)


        c = ConnectedDatabase.cursor()

        c.execute('PRAGMA TABLE_INFO({})'.format(table_name))

        table_header = c.fetchall()

        ConnectedDatabase.commit()
        ConnectedDatabase.close()

        return table_header  # zero indexed, LIST class of tuple of 5 elements.

    except IOError:
        return None


def check_header_index(database_path, table_name, field):
    """
    Parse the list of the header and then check it against the field provided to return the index.
    :param database_path:
    :param table_name:
    :param field:
    :return:
    """

    try:
        header_list = check_header(database_path, table_name)  # header list is a list of 5 elements tuples.


        if header_list is not None:
            global header_index

            for table_column in header_list:
                if table_column[1] == field:      # 1 is field name.
                    global header_index

                    header_index = table_column[0]    # 0 is the index
                    break
            return header_index
    except IOError:
        return None

def validateLocalTableAndSchema(database_path, table_name, field):

    """
    Does a comprehensitve check to ensure the field, 1) exist in the schema, 2) exist in the local table and then are the SAME!

    :param database_path:
    :param table_name:
    :param field: the string of the field name that is to be searched.
    :return:
    """
    logger = logging.getLogger('LORISQuery_validateLocalTableAndSchema')
    field_table_index = -1
    field_schema_index = -2


    # Schema check: note that schema contains keyfield
    if field not in CNBP_schema:
        return False, "Current planned schema does not contain " + field
    else:
        field_schema_index = CNBP_schema.index(field)

    # Table header check: table also must contain keyfield
    try:
        field_table_index = check_header_index(database_path, table_name, field)
        if field_table_index is None:
            return False, "SQLite table HEADER does not contain " + field
        else:
            logger.info("SQLite table HEADER for the field " + field + " is " + str(field_table_index))

    except IOError:
        return False, "Database not reachable"

    if field_table_index < 0 or field_schema_index < 0:
        return False, "Check program for bugs. Default values not modified"

    # This ensure we checking the right column for field information by validating against both schema and table.

    if field_table_index != field_schema_index:
        return False, "Schema & Table definition not matching"

    return True, "Database and Schema congruently support this field and its position"


#if __name__ == '__main__':




1			import sys
			0 ignored issues – show Coding Style introduced 2018-07-31 12:18 UTC by Report Bug Copy Issue Report This module should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
2			import logging
3			import sqlite3
4			from pathlib import Path
5			from LocalDB.schema import CNBP_schema
			0 ignored issues – show Bug introduced 2018-08-01 16:25 UTC by Report Bug Copy Issue Report The name `schema` does not seem to exist in module `LocalDB`. Loading history... introduced 2018-08-01 16:25 UTC by Report Bug Copy Issue Report Unable to import 'LocalDB.schema' Loading history...
6
7			logging.basicConfig(stream=sys.stdout, level=logging.INFO)
8
9
10			def check_value(database_path, table_name, ColumnName, ColumnValue):
			0 ignored issues – show Coding Style Naming introduced 2018-07-31 12:18 UTC by Report Bug Copy Issue Report The name `ColumnName` does not conform to the argument naming conventions (`(([a-z][a-z0-9_]{2,30})\|(_[a-z0-9_]))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... Coding Style Naming introduced 2018-07-31 12:18 UTC by Report Bug Copy Issue Report The name `ColumnValue` does not conform to the argument naming conventions (`(([a-z][a-z0-9_]{2,30})\|(_[a-z0-9_]))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
11			"""
12			Check if a subject exist in the given database and given table
13			:param database_path: path to the SQLite database
14			:param table_name: the name of the table being queried
15			:param ColumnName: the column being queried
16			:param ColumnValue: the value of the column being checked
17			:return: boolean on if this is ever found in the given database in the given table, in the given column.
			0 ignored issues – show Coding Style introduced 2018-07-31 12:18 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (108/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
18			"""
19			logger = logging.getLogger('LORISQuery_CheckSubjectExist')
20
21
22			SQLPath = Path(database_path)
			0 ignored issues – show Coding Style Naming introduced 2018-07-31 12:18 UTC by Report Bug Copy Issue Report The name `SQLPath` does not conform to the variable naming conventions (`(([a-z][a-z0-9_]{2,30})\|(_[a-z0-9_]*))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
23
24			# check if path is a file and exist.
25			if not SQLPath.is_file():
26			logger.info('SQLite database file does not exist!')
27			return False
28
29			# Try to connect the database to start the process:
30			try:
31			# Create on Connecting to the database file
32			ConnectedDatabase = sqlite3.connect(database_path)
			0 ignored issues – show Coding Style Naming introduced 2018-07-31 12:18 UTC by Report Bug Copy Issue Report The name `ConnectedDatabase` does not conform to the variable naming conventions (`(([a-z][a-z0-9_]{2,30})\|(_[a-z0-9_]*))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
33			c = ConnectedDatabase.cursor()
			0 ignored issues – show Coding Style Naming introduced 2018-07-31 12:18 UTC by Report Bug Copy Issue Report The name `c` does not conform to the variable naming conventions (`(([a-z][a-z0-9_]{2,30})\|(_[a-z0-9_]*))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...

CNBP / DICOMTransit

Pull Request — master (#2)

Python.LocalDB.query.check_value() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like