KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use update_schema_file_and_node method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

osmosis.py

Source:osmosis.py

...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:1150                    model_column.update(prior_knowledge)1151            changes_committed += 11152            logger().info(1153                ":light_bulb: Column %s is inheriting knowledge from the lineage of progenitor (%s)",1154                column,1155                progenitor,1156            )1157            logger().info(prior_knowledge)1158        return changes_committed1159    @staticmethod1160    def add_missing_cols_to_node_and_model(1161        missing_columns: Iterable,1162        node: ManifestNode,1163        yaml_file_model_section: Dict[str, Any],1164    ) -> int:1165        """Add missing columns to node and model simultaneously1166        THIS MUTATES THE NODE AND MODEL OBJECTS so that state is always accurate"""1167        changes_committed = 01168        for column in missing_columns:1169            node.columns[column] = ColumnInfo.from_dict({"name": column})1170            yaml_file_model_section.setdefault("columns", []).append({"name": column})1171            changes_committed += 11172            logger().info(":syringe: Injecting column %s into dbt schema", column)1173        return changes_committed1174    def update_schema_file_and_node(1175        self,1176        missing_columns: Iterable[str],1177        undocumented_columns: Iterable[str],1178        extra_columns: Iterable[str],1179        node: ManifestNode,1180        yaml_file: Dict[str, Any],1181    ) -> Tuple[int, int, int]:1182        """Take action on a schema file mirroring changes in the node."""1183        # We can extrapolate this to a general func1184        noop = 0, 0, 01185        if node.resource_type == NodeType.Source:1186            KEY = "tables"1187            yaml_file_models = None1188            for src in yaml_file.get("sources", []):...

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.