Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use pretty_print_restructure_plan method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

osmosis.py

Source:osmosis.py

...889        if not blueprint:890            logger().info(":1st_place_medal: Project structure approved")891            return False892        # Print plan for user auditability893        self.pretty_print_restructure_plan(blueprint)894        logger().info(895            ":construction_worker: Executing action plan and conforming projecting schemas to defined structure"896        )897        for target, structure in blueprint.items():898            if not target.exists():899                # Build File900                logger().info(":construction: Building schema file %s", target.name)901                if not self.dry_run:902                    target.parent.mkdir(exist_ok=True, parents=True)903                    target.touch()904                    self.yaml_handler.dump(structure.output, target)905            else:906                # Update File907                logger().info(":toolbox: Updating schema file %s", target.name)908                target_schema: Optional[Dict[str, Any]] = self.yaml_handler.load(target)909                if not target_schema:910                    target_schema = {"version": 2}911                elif "version" not in target_schema:912                    target_schema["version"] = 2913                target_schema.setdefault("models", []).extend(structure.output["models"])914                if not self.dry_run:915                    self.yaml_handler.dump(target_schema, target)916            # Clean superseded schema files917            for dir, models in structure.supersede.items():918                preserved_models = []919                raw_schema: Dict[str, Any] = self.yaml_handler.load(dir)920                models_marked_for_superseding = set(models)921                models_in_schema = set(map(lambda mdl: mdl["name"], raw_schema.get("models", [])))922                non_superseded_models = models_in_schema - models_marked_for_superseding923                if len(non_superseded_models) == 0:924                    logger().info(":rocket: Superseded schema file %s", dir.name)925                    if not self.dry_run:926                        dir.unlink(missing_ok=True)927                else:928                    for model in raw_schema["models"]:929                        if model["name"] in non_superseded_models:930                            preserved_models.append(model)931                    raw_schema["models"] = preserved_models932                    if not self.dry_run:933                        self.yaml_handler.dump(raw_schema, dir)934                    logger().info(935                        ":satellite: Model documentation migrated from %s to %s",936                        dir.name,937                        target.name,938                    )939        return True940    @staticmethod941    def pretty_print_restructure_plan(blueprint: Dict[Path, SchemaFileMigration]) -> None:942        logger().info(943            list(944                map(945                    lambda plan: (blueprint[plan].supersede or "CREATE", "->", plan),946                    blueprint.keys(),947                )948            )949        )950    def build_node_ancestor_tree(951        self,952        node: ManifestNode,953        family_tree: Optional[Dict[str, List[str]]] = None,954        members_found: Optional[List[str]] = None,955        depth: int = 0,...

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.