DbProfilerRepository.py 文件源码

def get_datamap_source_tables(self, database_name, schema_name,
                                  table_name):
        """Get source talbe names from the data mapping info.

        Args:
          database_name (str):
          schema_name (str):
          tablename (str):

        Returns:
          list: a list of source table names.
        """
        assert database_name and schema_name and table_name

        tables = []
        for d in self.get_datamap_items(database_name, schema_name,
                                        table_name):
            if d['source_table_name'] not in tables:
                tables.append(d['source_table_name'])

        return tables