pystratum_pgsql.backend.PgSqlConstantWorker

Complexity

Total Complexity

44

Size/Duplication

Total Lines	235
Duplicated Lines	0 %

Test Coverage

Coverage

92.59%

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
B	PgSqlConstantWorker._enhance_columns()	0	19	6
A	PgSqlConstantWorker._get_labels()	0	9	3
A	PgSqlConstantWorker.derive_field_length()	0	34	2
A	PgSqlConstantWorker._read_configuration_file()	0	5	1
B	PgSqlConstantWorker._write_columns()	0	34	6
A	PgSqlConstantWorker.__init__()	0	11	1
C	PgSqlConstantWorker._get_old_columns()	0	41	10
A	PgSqlConstantWorker._get_columns()	0	16	4
A	PgSqlConstantWorker._fill_constants()	0	11	5
B	PgSqlConstantWorker._merge_columns()	0	14	6

import os
import re
from configparser import ConfigParser
from typing import Any, Dict, Optional

from pystratum_backend.StratumStyle import StratumStyle
from pystratum_common.backend.CommonConstantWorker import CommonConstantWorker
from pystratum_common.Util import Util

from pystratum_pgsql.backend.PgSqlWorker import PgSqlWorker


class PgSqlConstantWorker(PgSqlWorker, CommonConstantWorker):
    """
    Class for creating constants based on column widths, and auto increment columns and labels for PostgreSQL databases.
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self, io: StratumStyle, config: ConfigParser):
        """
        Object constructor.

        :param PyStratumStyle io: The output decorator.
        """
        PgSqlWorker.__init__(self, io, config)
        CommonConstantWorker.__init__(self, io, config)

        self._columns: Dict[str, Any] = {}
        """
        Metadata of all columns in the current schema and information_schema.

        :type: dict
        """

    # ------------------------------------------------------------------------------------------------------------------
    def _get_old_columns(self) -> None:
        """
        Reads from file constants_filename the previous table and column names, the width of the column,
        and the constant name (if assigned) and stores this data in old_columns.
        """
        if os.path.exists(self._constants_filename):
            with open(self._constants_filename, 'r') as stream:
                for line in stream:
                    if line != "\n":
                        prog = re.compile(r'\s*(?:([a-zA-Z0-9_]+)\.)?([a-zA-Z0-9_]+)\.'
                                          r'([a-zA-Z0-9_]+)\s+(\d+)\s*(\*|[a-zA-Z0-9_]+)?\s*')
                        matches = prog.findall(line)

                        if matches:
                            matches = matches[0]
                            schema_name = str(matches[0])
                            table_name = str(matches[1])
                            column_name = str(matches[2])
                            length = str(matches[3])
                            constant_name = str(matches[4])

                            if schema_name:
                                table_name = schema_name + '.' + table_name

                            if constant_name:
                                column_info = {'table_name':    table_name,
                                               'column_name':   column_name,
                                               'length':        length,
                                               'constant_name': constant_name}
                            else:
                                column_info = {'table_name':  table_name,
                                               'column_name': column_name,
                                               'length':      length}

                            if table_name in self._old_columns:
                                if column_name in self._old_columns[table_name]:
                                    pass
                                else:
                                    self._old_columns[table_name][column_name] = column_info
                            else:
                                self._old_columns[table_name] = {column_name: column_info}

    # ------------------------------------------------------------------------------------------------------------------
    def _get_columns(self) -> None:
        """
        Retrieves metadata all columns in the current schema and information_schema.
        """
        rows = self._dl.get_all_table_columns()
        for row in rows:
            # Enhance row with the actual length of the column.
            row['length'] = self.derive_field_length(row)

            if row['table_name'] in self._columns:
                if row['column_name'] in self._columns[row['table_name']]:
                    pass
                else:
                    self._columns[row['table_name']][row['column_name']] = row
            else:
                self._columns[row['table_name']] = {row['column_name']: row}

    # ------------------------------------------------------------------------------------------------------------------
    def _enhance_columns(self) -> None:
        """
        Enhances old_columns as follows:
        If the constant name is *, is is replaced with the column name prefixed by prefix in uppercase.
        Otherwise the constant name is set to uppercase.
        """
        if self._old_columns:
            for table_name, table in sorted(self._old_columns.items()):
                for column_name, column in sorted(table.items()):
                    table_name = column['table_name']
                    column_name = column['column_name']

                    if 'constant_name' in column:
                        if column['constant_name'].strip() == '*':
                            constant_name = str(self._prefix + column['column_name']).upper()
                            self._old_columns[table_name][column_name]['constant_name'] = constant_name
                        else:
                            constant_name = str(self._old_columns[table_name][column_name]['constant_name']).upper()
                            self._old_columns[table_name][column_name]['constant_name'] = constant_name

    # ------------------------------------------------------------------------------------------------------------------
    def _merge_columns(self) -> None:
        """
        Preserves relevant data in old_columns into columns.
        """
        if self._old_columns:
            for table_name, table in sorted(self._old_columns.items()):
                for column_name, column in sorted(table.items()):
                    if 'constant_name' in column:
                        try:
                            self._columns[table_name][column_name]['constant_name'] = column['constant_name']
                        except KeyError:
                            # Either the column or table is not present anymore.
                            self._io.warning('Dropping constant {0} because column is not present anymore'.
                                             format(column['constant_name']))

    # ------------------------------------------------------------------------------------------------------------------
    def _write_columns(self) -> None:
        """
        Writes table and column names, the width of the column, and the constant name (if assigned) to
        constants_filename.
        """
        content = ''
        for _, table in sorted(self._columns.items()):
            width1 = 0
            width2 = 0

            key_map = {}
            for column_name, column in table.items():
                key_map[column['ordinal_position']] = column_name
                width1 = max(len(str(column['column_name'])), width1)
                width2 = max(len(str(column['length'])), width2)

            for _, column_name in sorted(key_map.items()):
                if table[column_name]['length'] is not None:
                    if 'constant_name' in table[column_name]:
                        line_format = "%s.%-{0:d}s %{1:d}d %s\n".format(int(width1), int(width2))
                        content += line_format % (table[column_name]['table_name'],
                                                  table[column_name]['column_name'],
                                                  table[column_name]['length'],
                                                  table[column_name]['constant_name'])
                    else:
                        line_format = "%s.%-{0:d}s %{1:d}d\n".format(int(width1), int(width2))
                        content += line_format % (table[column_name]['table_name'],
                                                  table[column_name]['column_name'],
                                                  table[column_name]['length'])

            content += "\n"

        # Save the columns, width and constants to the filesystem.
        Util.write_two_phases(self._constants_filename, content, self._io)

    # ------------------------------------------------------------------------------------------------------------------
    def _get_labels(self) -> None:
        """
        Gets all primary key labels from the current schema.
        """
        tables = self._dl.get_label_tables(self._label_regex)
        for table in tables:
            rows = self._dl.get_labels_from_table(table['table_name'], table['id'], table['label'])
            for row in rows:
                self._labels[row['label']] = row['id']

    # ------------------------------------------------------------------------------------------------------------------
    def _fill_constants(self) -> None:
        """
        Merges columns and labels (i.e. all known constants) into constants.
        """
        for _, table in sorted(self._columns.items()):
            for _, column in sorted(table.items()):
                if 'constant_name' in column:
                    self._constants[column['constant_name']] = column['length']

        for label, label_id in sorted(self._labels.items()):
            self._constants[label] = label_id

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def derive_field_length(column: Dict[str, Any]) -> Optional[int]:
        """
        Returns the width of a field based on the data type of column.

        :param dict column: The column of which the field is based.

        :rtype: int|None
        """
        types_length = {'bigint':                      21,
                        'integer':                     11,
                        'smallint':                    6,
                        'bit':                         column['character_maximum_length'],
                        'money':                       None,
                        'boolean':                     None,
                        'double':                      column['numeric_precision'],
                        'numeric':                     column['numeric_precision'],
                        'real':                        None,
                        'character':                   column['character_maximum_length'],
                        'character varying':           column['character_maximum_length'],
                        'point':                       None,
                        'polygon':                     None,
                        'text':                        None,
                        'bytea':                       None,
                        'xml':                         None,
                        'USER-DEFINED':                None,
                        'timestamp without time zone': 16,
                        'time without time zone':      8,
                        'date':                        10}

        if column['data_type'] in types_length:
            return types_length[column['data_type']]

        raise Exception("Unexpected type '{0!s}'.".format(column['data_type']))

    # ------------------------------------------------------------------------------------------------------------------
    def _read_configuration_file(self) -> None:
        """
        Reads parameters from the configuration file.
        """
        CommonConstantWorker._read_configuration_file(self)

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