pystratum_mysql.wrapper.MySqlWrapper

Complexity

Total Complexity

13

Size/Duplication

Total Lines	137
Duplicated Lines	0 %

Test Coverage

Coverage

20%

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	MySqlWrapper.is_lob_parameter()	0	46	4
A	MySqlWrapper._get_parameter_format_specifier()	0	40	2
B	MySqlWrapper._generate_command()	0	34	7

from abc import ABC
from typing import Any, Dict, List

from pystratum_common.wrapper.Wrapper import Wrapper


class MySqlWrapper(Wrapper, ABC):
    """
    Parent class for wrapper method generators for stored procedures and functions.
    """

    # ------------------------------------------------------------------------------------------------------------------
    def is_lob_parameter(self, parameters: List[Dict[str, Any]]) -> bool:
        """
        Returns True of one of the parameters is a BLOB or CLOB. Otherwise, returns False.

        :param parameters: The parameters of a stored routine.
        """
        has_lob = False

        lookup = {'bigint':     False,
                  'binary':     False,
                  'bit':        False,
                  'char':       False,
                  'date':       False,
                  'datetime':   False,
                  'decimal':    False,
                  'double':     False,
                  'enum':       False,
                  'float':      False,
                  'int':        False,
                  'mediumint':  False,
                  'set':        False,
                  'smallint':   False,
                  'time':       False,
                  'timestamp':  False,
                  'tinyint':    False,
                  'varbinary':  False,
                  'varchar':    False,
                  'year':       False,

                  'blob':       True,
                  'longblob':   True,
                  'longtext':   True,
                  'mediumblob': True,
                  'mediumtext': True,
                  'text':       True,
                  'tinyblob':   True,
                  'tinytext':   True}

        if parameters:
            for parameter_info in parameters:
                if parameter_info['data_type'] in lookup:
                    has_lob = lookup[parameter_info['data_type']]
                else:
                    raise Exception("Unexpected date type '{0!s}'.".format(parameter_info['data_type']))

        return has_lob

    # ------------------------------------------------------------------------------------------------------------------
    def _generate_command(self, routine: Dict[str, Any]) -> str:
        """
        Generates SQL statement for calling a stored routine.

        :param routine: Metadata of the stored routine.
        """
        parameters = ''
        placeholders = ''

        execute = 'call'
        if routine['designation'] == 'function':
            execute = 'select'

        i = 0
        l = 0
        for parameter in routine['parameters']:
            re_type = self._get_parameter_format_specifier(parameter['data_type'])
            if parameters:
                parameters += ', '
                placeholders += ', '
            parameters += parameter['name']
            placeholders += re_type
            i += 1
            if not re_type == '?':
                l += 1

        if l == 0:
            line = '"{0!s} {1!s}()"'.format(execute, routine['routine_name'])
        elif l >= 1:
            line = '"{0!s} {1!s}({2!s})", {3!s}'.format(execute, routine['routine_name'], placeholders, parameters)
        else:
            raise Exception('Internal error.')

        return line

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def _get_parameter_format_specifier(data_type: str) -> str:
        """
        Returns the appropriate format specifier for a parameter type.

        :param data_type: The parameter type.
        """
        lookup = {'bigint':     '%s',
                  'binary':     '%s',
                  'bit':        '%s',
                  'blob':       '%s',
                  'char':       '%s',
                  'date':       '%s',
                  'datetime':   '%s',
                  'decimal':    '%s',
                  'double':     '%s',
                  'enum':       '%s',
                  'float':      '%s',
                  'int':        '%s',
                  'longblob':   '%s',
                  'longtext':   '%s',
                  'mediumblob': '%s',
                  'mediumint':  '%s',
                  'mediumtext': '%s',
                  'set':        '%s',
                  'smallint':   '%s',
                  'text':       '%s',
                  'time':       '%s',
                  'timestamp':  '%s',
                  'tinyblob':   '%s',
                  'tinyint':    '%s',
                  'tinytext':   '%s',
                  'varbinary':  '%s',
                  'varchar':    '%s',
                  'year':       '%s'}

        if data_type in lookup:
            return lookup[data_type]

        raise Exception('Unexpected data type {0!s}.'.format(data_type))

# ----------------------------------------------------------------------------------------------------------------------