How to use remove_columns_not_in_database method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

osmosis.py

Source:osmosis.py

...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", []):1189                if src["name"] == node.source_name:1190                    # Scope our pointer to a specific portion of the object1191                    yaml_file_models = src1192        else:1193            KEY = "models"1194            yaml_file_models = yaml_file1195        if yaml_file_models is None:1196            return noop1197        for yaml_file_model_section in yaml_file_models[KEY]:1198            if yaml_file_model_section["name"] == node.name:1199                logger().info(":microscope: Looking for actions")1200                n_cols_added = self.add_missing_cols_to_node_and_model(1201                    missing_columns, node, yaml_file_model_section1202                )1203                n_cols_doc_inherited = self.update_undocumented_columns_with_prior_knowledge(1204                    undocumented_columns, node, yaml_file_model_section1205                )1206                n_cols_removed = self.remove_columns_not_in_database(1207                    extra_columns, node, yaml_file_model_section1208                )1209                return n_cols_added, n_cols_doc_inherited, n_cols_removed1210        logger().info(":thumbs_up: No actions needed")...

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.