Extract DbtModel from a parsed manifest node.
(self, node_id: str, node: Any)
| 126 | ) |
| 127 | |
| 128 | def _extract_model_from_node(self, node_id: str, node: Any) -> Optional[DbtModel]: |
| 129 | """Extract DbtModel from a parsed manifest node.""" |
| 130 | # Check resource type |
| 131 | resource_type = getattr(node, "resource_type", None) |
| 132 | if resource_type is None: |
| 133 | if not node_id.startswith("model."): |
| 134 | return None |
| 135 | else: |
| 136 | resource_type_str = ( |
| 137 | resource_type.value |
| 138 | if hasattr(resource_type, "value") |
| 139 | else str(resource_type) |
| 140 | ) |
| 141 | if resource_type_str != "model": |
| 142 | return None |
| 143 | |
| 144 | model_name = getattr(node, "name", "") |
| 145 | node_tags = list(getattr(node, "tags", []) or []) |
| 146 | node_columns = getattr(node, "columns", {}) or {} |
| 147 | depends_on = getattr(node, "depends_on", None) |
| 148 | |
| 149 | if depends_on: |
| 150 | depends_on_nodes = list(getattr(depends_on, "nodes", []) or []) |
| 151 | else: |
| 152 | depends_on_nodes = [] |
| 153 | |
| 154 | # Extract columns |
| 155 | columns = [ |
| 156 | self._extract_column_from_node(col_name, col_data) |
| 157 | for col_name, col_data in node_columns.items() |
| 158 | ] |
| 159 | |
| 160 | # Get schema - dbt-artifacts-parser uses schema_ to avoid Python keyword |
| 161 | schema = getattr(node, "schema_", "") or getattr(node, "schema", "") or "" |
| 162 | |
| 163 | return DbtModel( |
| 164 | name=model_name, |
| 165 | unique_id=node_id, |
| 166 | database=getattr(node, "database", "") or "", |
| 167 | schema=schema, |
| 168 | alias=getattr(node, "alias", model_name) or model_name, |
| 169 | description=getattr(node, "description", "") or "", |
| 170 | columns=columns, |
| 171 | tags=node_tags, |
| 172 | meta=dict(getattr(node, "meta", {}) or {}), |
| 173 | depends_on=depends_on_nodes, |
| 174 | ) |
| 175 | |
| 176 | def get_models( |
| 177 | self, |
no test coverage detected