How to use get_node_columns_with_inherited_knowledge method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

osmosis.py

Source:osmosis.py

...1002                    if not deserialized_info["meta"]:1003                        deserialized_info.pop("meta")1004                    knowledge[name].update(deserialized_info)1005        return knowledge1006    def get_node_columns_with_inherited_knowledge(1007        self,1008        node: ManifestNode,1009    ) -> Dict[str, Dict[str, Any]]:1010        """Build a knowledgebase for the model based on iterating through ancestors"""1011        family_tree = self.build_node_ancestor_tree(node)1012        knowledge = self.inherit_column_level_knowledge(family_tree)1013        return knowledge1014    @staticmethod1015    def get_column_sets(1016        database_columns: Iterable[str],1017        yaml_columns: Iterable[str],1018        documented_columns: Iterable[str],1019    ) -> Tuple[List[str], List[str], List[str]]:1020        """Returns:1021        missing_columns: Columns in database not in dbt -- will be injected into schema file1022        undocumented_columns: Columns missing documentation -- descriptions will be inherited and injected into schema file where prior knowledge exists1023        extra_columns: Columns in schema file not in database -- will be removed from schema file1024        """1025        missing_columns = [1026            x for x in database_columns if x.lower() not in (y.lower() for y in yaml_columns)1027        ]1028        undocumented_columns = [1029            x for x in database_columns if x.lower() not in (y.lower() for y in documented_columns)1030        ]1031        extra_columns = [1032            x for x in yaml_columns if x.lower() not in (y.lower() for y in database_columns)1033        ]1034        return missing_columns, undocumented_columns, extra_columns1035    def propagate_documentation_downstream(self, force_inheritance: bool = False) -> None:1036        schema_map = self.build_schema_folder_mapping()1037        with self.adapter.connection_named("dbt-osmosis"):1038            for unique_id, node in track(list(self.filtered_models())):1039                logger().info("\n:point_right: Processing model: [bold]%s[/bold] \n", unique_id)1040                # Get schema file path, must exist to propagate documentation1041                schema_path: Optional[SchemaFileLocation] = schema_map.get(unique_id)1042                if schema_path is None or schema_path.current is None:1043                    logger().info(1044                        ":bow: No valid schema file found for model %s", unique_id1045                    )  # We can't take action1046                    continue1047                # Build Sets1048                database_columns: Set[str] = set(self.get_columns(node))1049                yaml_columns: Set[str] = set(column for column in node.columns)1050                if not database_columns:1051                    logger().info(1052                        ":safety_vest: Unable to resolve columns in database, falling back to using yaml columns as base column set\n"1053                    )1054                    database_columns = yaml_columns1055                # Get documentated columns1056                documented_columns: Set[str] = set(1057                    column1058                    for column, info in node.columns.items()1059                    if info.description and info.description not in self.placeholders1060                )1061                # Queue1062                missing_columns, undocumented_columns, extra_columns = self.get_column_sets(1063                    database_columns, yaml_columns, documented_columns1064                )1065                if force_inheritance:1066                    # Consider all columns "undocumented" so that inheritance is not selective1067                    undocumented_columns = database_columns1068                # Engage1069                n_cols_added = 01070                n_cols_doc_inherited = 01071                n_cols_removed = 01072                if len(missing_columns) > 0 or len(undocumented_columns) or len(extra_columns) > 0:1073                    schema_file = self.yaml_handler.load(schema_path.current)1074                    (1075                        n_cols_added,1076                        n_cols_doc_inherited,1077                        n_cols_removed,1078                    ) = self.update_schema_file_and_node(1079                        missing_columns,1080                        undocumented_columns,1081                        extra_columns,1082                        node,1083                        schema_file,1084                    )1085                    if n_cols_added + n_cols_doc_inherited + n_cols_removed > 0:1086                        # Dump the mutated schema file back to the disk1087                        if not self.dry_run:1088                            self.yaml_handler.dump(schema_file, schema_path.current)1089                        logger().info(":sparkles: Schema file updated")1090                # Print Audit Report1091                n_cols = float(len(database_columns))1092                n_cols_documented = float(len(documented_columns)) + n_cols_doc_inherited1093                perc_coverage = (1094                    min(100.0 * round(n_cols_documented / n_cols, 3), 100.0)1095                    if n_cols > 01096                    else "Unable to Determine"1097                )1098                logger().info(1099                    self.audit_report.format(1100                        database=node.database,1101                        schema=node.schema,1102                        table=node.name,1103                        total_columns=n_cols,1104                        n_cols_added=n_cols_added,1105                        n_cols_doc_inherited=n_cols_doc_inherited,1106                        n_cols_removed=n_cols_removed,1107                        coverage=perc_coverage,1108                    )1109                )1110    @staticmethod1111    def remove_columns_not_in_database(1112        extra_columns: Iterable[str],1113        node: ManifestNode,1114        yaml_file_model_section: Dict[str, Any],1115    ) -> int:1116        """Removes columns found in dbt model that do not exist in database from both node and model simultaneously1117        THIS MUTATES THE NODE AND MODEL OBJECTS so that state is always accurate"""1118        changes_committed = 01119        for column in extra_columns:1120            node.columns.pop(column, None)1121            yaml_file_model_section["columns"] = [1122                c for c in yaml_file_model_section["columns"] if c["name"] != column1123            ]1124            changes_committed += 11125            logger().info(":wrench: Removing column %s from dbt schema", column)1126        return changes_committed1127    def update_undocumented_columns_with_prior_knowledge(1128        self,1129        undocumented_columns: Iterable[str],1130        node: ManifestNode,1131        yaml_file_model_section: Dict[str, Any],1132    ) -> int:1133        """Update undocumented columns with prior knowledge in node and model simultaneously1134        THIS MUTATES THE NODE AND MODEL OBJECTS so that state is always accurate"""1135        knowledge = self.get_node_columns_with_inherited_knowledge(node)1136        inheritables = ("description", "tags", "meta")1137        changes_committed = 01138        for column in undocumented_columns:1139            prior_knowledge = knowledge.get(column, {})1140            progenitor = prior_knowledge.pop("progenitor", "Unknown")1141            prior_knowledge = {k: v for k, v in prior_knowledge.items() if k in inheritables}1142            if not prior_knowledge:1143                continue1144            if column not in node.columns:1145                node.columns[column] = ColumnInfo.from_dict({"name": column, **prior_knowledge})1146            else:1147                node.columns[column].replace(kwargs={"name": column, **prior_knowledge})1148            for model_column in yaml_file_model_section["columns"]:1149                if model_column["name"] == column:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.