cdplib/cdplib/db_migration/ParseMapping.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Fri Sep 20 15:33:17 2019

@author: tanya
"""

import os
import sys
import numpy as np
import json

from cdplib.log import Log
sys.path.append(os.getcwd())

class ParseMapping:
    '''
    '''
    def __init__(self, mapping_paths: (str, list), log_name: str = "ParseMapping",
                 source: str = "original_name", target: str = "mongo_name",
                 target_collections: str = "mongo_collection"):
        '''
        '''

        self._log = Log('Parse Mapping')

        assert(isinstance(mapping_paths, (list, str))),\
            "Mapping_paths must be either str or lists"

        if isinstance(mapping_paths, str):
            mapping_paths = [mapping_paths]

        self._mapping_paths =  mapping_paths
        self._source = source
        self._target = target
        self._target_collections = target_collections
        self._update_mapping()

    def _update_mapping(self):
        '''
        Since we can have multiple mappings per table we need to add them to
        the object. I concatenated the mapping so that we don't have to adjust
        all function of the class to accept also list input. The class could
        be adjusted to accept list or even a dictornary with the key name as
        name of the mapping and value the json mapping.
        !!! WARNING !!!!
        Since the mapping are just concatenated there is right now
        no way to ditinguish from the object itself which item belongs to which
        mapping file.
        '''
        mappings = []

        for mapping_path in self._mapping_paths:
            try:
                with open(mapping_path, "r") as f:
                    mapping = json.load(f)
                mappings.append(mapping)

            except Exception as e:
                err = ("Could not load json schema:{1} , "
                       "Obtained error {0}".format(e, mapping_path))

                self._log.error(err)
                raise Exception(err)

        if len(mappings) > 1:
            concatenate_mapping = []
            for mapping in mappings:
                if not concatenate_mapping:
                    concatenate_mapping = mapping
                else:
                   concatenate_mapping.extend(mapping)
            self._mapping = concatenate_mapping
        else:
            self._mapping = mappings[0]


    def get_field_mapping(self) -> dict:
        '''
        '''
        assert(all([set([self._source, self._target]) <= set(d)
                    for d in self._mapping]))

        return {d[self._source]: d[self._target] for d in self._mapping}

    def _get_fields_satistisfying_condition(self, key: str, value) -> list:
        '''
        '''
        assert(all([self._source in d for d in self._mapping])),\
            "Invalid from field"

        return [d[self._source] for d in self._mapping
                if (key in d) and (value in d[key])]

    def get_required_fields(self) -> list:
        '''
        '''
        return self._get_fields_satistisfying_condition(key="required",
                                                        value=1)

    def get_date_fields(self) -> list:
        '''
        '''
        return self._get_fields_satistisfying_condition(key="type",
                                                        value="Date")

    def get_fields_restricted_to_collection(self, collection_name: str) -> list:
        '''
        '''
        return self._get_fields_satistisfying_condition(key=self._target_collections,
                                                        value=collection_name)

    def _get_property_from_mapping(self, property_names: list) -> dict:
        '''
        Get specified property names from migration mapping json.
        '''
        assert(isinstance(property_names,list)),\
            "Parameter 'property_names' is not a list"

        assert(all([self._source in d for d in self._mapping])),\
            "Not all objects in the mapping json contain property tag " + self._source

        result = {}
        for column_mapping in self._mapping:
            for property_name in property_names:
                if property_name in column_mapping and column_mapping[property_name]:
                 result.update({column_mapping[self._source]: column_mapping[property_name]})

        return result



    def get_default_values(self) -> dict:
        '''
        Get default values from migration mapping json. If more peorerty names
        are beeing added also add them in the unit test.
        '''
        standard_default_names=["default_values"]

        return self._get_property_from_mapping(standard_default_names)

    def get_types(self) -> dict:
        '''
        Get type from migration mapping json. If more peorerty names
        are beeing added also add them in the unit test.
        '''

        standard_type_names=["type"]

        return self._get_property_from_mapping(standard_type_names)

    def get_value_mappings(self) -> dict:
        '''
        Get type from migration mapping json. If more peorerty names
        are beeing added also add them in the unit test.
        '''

        standard_value_mapping_names = ["value_mapping"]

        return self._get_property_from_mapping(standard_value_mapping_names)

    def get_date_formats(self) -> dict:
        '''
        Get date fromats from migration mapping json. If more peorerty names
        are beeing added or value also add them in the unit test.
        '''
        
        standard_date_format_names = ["date_format"]

        return self._get_property_from_mapping(standard_date_format_names)
    
    def get_internal_names(self) -> dict:
        '''
        '''

        if all(["internal_name" in d for d in self._mapping]):
            internal_names = [d["internal_name"] for d in self._mapping]

        elif all(["internal_name" not in d for d in self._mapping]):
            internal_names = list(range(len(self._mapping)))


        else:
            err = ("Incorrectly filled mapping. Internal names should "
                   "either be in all or in neither of the fields")
            self._log.error(err)
            raise Exception(err)

        return internal_names

    def get_mongo_names(self) -> dict:
        '''
        '''
        if all(["mongo_name" in d for d in self._mapping]):
            mongo_names = [d["mongo_name"] for d in self._mapping]

        elif all(["mongo_name" not in d for d in self._mapping]):
            mongo_names = list(range(len(self._mapping)))

        else:
            err = ("Incorrectly filled mapping. Mongo names should "
                   "either be in all or in neither of the fields")
            self._log.error(err)
            raise Exception(err)

        return mongo_names



    def get_python_types(self) -> dict:
        '''
        '''
        sql_to_python_dtypes = {
                "Text": str,
                "Date": np.dtype('<M8[ns]'),
                "Double": float,
                "Integer": int
                }

        sql_types = self.get_types()

        return {k: sql_to_python_dtypes[v] for k, v in sql_types.items()}



    def get_column_numbers(self) -> list:
        '''
        '''
        if all(["column_number" in d for d in self._mapping]):
            column_numbers = [d["column_number"] for d in self._mapping]

        elif all(["column_number" not in d for d in self._mapping]):
            column_numbers = list(range(len(self._mapping)))

        else:
            err = ("Incorrectly filled mapping. Column numbers should ",
                   "either in all or in neither of the fields")
            self._log.err(err)
            raise Exception(err)

        return column_numbers


if __name__ == "__main__":

    mapping_path = os.path.join(".", "migration_mappings", "unit_test_migration_mapping.json")

    if os.path.isfile(mapping_path):

        print("found mapping path")

        parser = ParseMapping(mapping_path, source="internal_name",
                              target="mongo_name")

        default_values = parser.get_default_values()
        print(default_values)
        date_formats = parser.get_date_formats()
        print(date_formats)
        mongo_names = parser.get_mongo_names()
        print(mongo_names)
        types = parser.get_types()
        print(types)
        column_numbers = parser.get_column_numbers()
        print(column_numbers)
        value_mappings = parser.get_value_mappings()
        print(value_mappings)
        date_formats = parser.get_date_formats()
        print(date_formats)

        print("Done testing!")