Extract dbt models from manifest. Args: model_names: Optional list of specific model names to extract tag_filter: Optional tag to filter models by Returns: List of DbtModel objects Example:: models = parser.get_mode
(
self,
model_names: Optional[List[str]] = None,
tag_filter: Optional[str] = None,
)
| 174 | ) |
| 175 | |
| 176 | def get_models( |
| 177 | self, |
| 178 | model_names: Optional[List[str]] = None, |
| 179 | tag_filter: Optional[str] = None, |
| 180 | ) -> List[DbtModel]: |
| 181 | """ |
| 182 | Extract dbt models from manifest. |
| 183 | |
| 184 | Args: |
| 185 | model_names: Optional list of specific model names to extract |
| 186 | tag_filter: Optional tag to filter models by |
| 187 | |
| 188 | Returns: |
| 189 | List of DbtModel objects |
| 190 | |
| 191 | Example:: |
| 192 | |
| 193 | models = parser.get_models(model_names=["driver_stats"]) |
| 194 | models = parser.get_models(tag_filter="feast") |
| 195 | """ |
| 196 | if self._parsed_manifest is None: |
| 197 | self.parse() |
| 198 | |
| 199 | if self._parsed_manifest is None: |
| 200 | return [] |
| 201 | |
| 202 | models = [] |
| 203 | nodes = getattr(self._parsed_manifest, "nodes", {}) or {} |
| 204 | |
| 205 | for node_id, node in nodes.items(): |
| 206 | # Only process models (not tests, seeds, snapshots, etc.) |
| 207 | if not node_id.startswith("model."): |
| 208 | continue |
| 209 | |
| 210 | model = self._extract_model_from_node(node_id, node) |
| 211 | if model is None: |
| 212 | continue |
| 213 | |
| 214 | # Filter by model names if specified |
| 215 | if model_names and model.name not in model_names: |
| 216 | continue |
| 217 | |
| 218 | # Filter by tag if specified |
| 219 | if tag_filter and tag_filter not in model.tags: |
| 220 | continue |
| 221 | |
| 222 | models.append(model) |
| 223 | |
| 224 | return models |
| 225 | |
| 226 | def get_model_by_name(self, model_name: str) -> Optional[DbtModel]: |
| 227 | """ |