source_to_sor_mappings.py 文件源码

def __init__(self, source: Union['SourceTable', 'SourceQuery', 'File'], target: Union[str, Table], auto_map: bool = True, filter='', ignore_fields: List[str] = []) -> None:
        #todo transformations
        if isinstance(source, File):
            self.file_name = source.file_name
        elif isinstance(source, SourceTable):
            self.source_table = source
        elif isinstance(source, SourceQuery):
            self.source_table = source

        if isinstance(target, str):
            self.sor_table = str(target) #type: str
        else:
            target_tbl = cast(Table, target)
            self.sor_table = target_tbl.name
        super().__init__(source, target, filter)
        self.temp_table = self.sor_table.replace('_hstage', '') + '_temp'
        #todo keys
        self.keys = [] #type: List[str]
        ignore_fields = [s.lower() for s in ignore_fields]
        self.ignore_fields = ignore_fields
        # self.field_mappings = [] #type: List[FieldMapping]
        self.auto_map = auto_map
        if auto_map: self.create_auto_mappings(source, ignore_fields)