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
sys.path.append(os.getcwd())

class ParseMapping:
    '''
    '''
    def __init__(self, mapping_path: str, log_name: str = "ParseMapping",
                 source_name_tag: str = "original_name", target_name_tag: str = "mongo_name",
                 target_collection_tag: str = "mongo_collection",
                 target_collection_name: str = None):
        '''
        '''
        import json
        from cdplib.log import Log

        self._log = Log('Parse Mapping')

        if not os.path.isfile(mapping_path):
            self._log.log_and_raise_error("Mapping not found")
        try:
            with open(mapping_path, "r") as f:
                self._mapping = json.load(f)

        except Exception as e:
            self._log.log_and_raise_error("Could not load mapping. "
                                          "Exit with error {}".format(e))

        self._mapping_path = mapping_path
        self._source_name_tag = source_name_tag
        self._target_name_tag = target_name_tag
        self._target_collection_tag = target_collection_tag
        self._target_collection_name = target_collection_name

        self._restrict_mapping_to_collection()

    def _restrict_mapping_to_collection(self):
        '''
        '''
        if self._target_collection_name is not None:

            result = []

            for d in self._mapping:

                if self._target_collection_tag not in d:
                    continue

                for key in [self._target_name_tag, self._target_collection_tag]:
                    if not isinstance(d[key], list):
                        d[key] = [d[key]]

                if (len(d[self._target_collection_tag]) > 1) and (len(d[self._target_name_tag]) == 1):
                    d[self._target_name_tag] = d[self._target_name_tag]*len(d[self._target_collection_tag])

                if len(d[self._target_collection_tag]) != len(d[self._target_name_tag]):
                    self._log.log_and_raise_error(("In the migration mapping '{0}' {1} "
                                                   "{2} has an unclear collection")
                                                   .format(self._mapping_path,
                                                           self._target_name_tag,
                                                           d[self._target_name_tag]))

                if self._target_collection_name in d[self._target_collection_tag]:

                    d[self._target_name_tag] = d[self._target_name_tag][d[self._target_collection_tag].index(self._target_collection_name)]

                    d[self._target_collection_tag] = self._target_collection_name

                    result.append(d)

            self._mapping = result

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

        return {d[self._source_name_tag]: d[self._target_name_tag] for d in self._mapping}

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

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

    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_collecton(self, collection_name: str) -> list:
        '''
        '''
        target_collection_tag_mapping = {d[self._source_name_tag]: d[self._target_collection_tag]
                                     for d in self._mapping
                                     if (self._target_collection_tag in d)}

        target_collection_tag_mapping = {k: v if isinstance(v, list) else [v]
                                     for k, v in target_collection_tag_mapping.items()}

        return [k for k,v in target_collection_tag_mapping.items()
                if collection_name in v]
    """

    def _get_info(self, key: str, value=None) -> dict:
        '''
        '''
        assert(all([self._source_name_tag in d for d in self._mapping])),\
            "Invalid from field"

        return {d[self._source_name_tag]: d[key] for d in self._mapping
                if (key in d) and ((value is not None)
                and (d[key] == value)) or (key in d)}

    def get_default_values(self) -> dict:
        '''
        '''
        return self._get_info(key="default_values")

    def get_date_formats(self) -> dict:
        '''
        '''
        return self._get_info(key="date_format")

    def get_types(self) -> dict:
        '''
        '''
        return self._get_info(key="type")

    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_value_mappings(self) -> dict:
        '''
        '''
        return self._get_info(key="value_mapping")

    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", "rs0_mapping.json")

    if os.path.isfile(mapping_path):

        print("found mapping path")

        parser = ParseMapping(mapping_path, source_name_tag="internal_name",
                              target_name_tag="mongo_name")

        internal_to_mongo_mapping = parser.get_field_mapping()

        original_to_internal_mapping = parser.get_field_mapping()

        default_values = parser.get_default_values()

        types = parser.get_types()

        column_numbers = parser.get_column_numbers()

        print("Done testing!")