from datetime import datetime, timedelta import json from typing import Optional, Dict, Any, List import warnings import yaml import pandas as pd import dagster as dg from dagster import AutomationCondition, AssetSpec, DagsterInvalidDefinitionError, apply_freshness_policy, map_asset_specs from dagster.preview.freshness import FreshnessPolicy from dagster._utils.schedules import is_valid_cron_string from dagster_dbt import ( DbtCliResource, DagsterDbtTranslator, DbtProject, dbt_assets ) from src.utils.dbt import DBT_MIN_PARTITION_DATE, get_dbt_tag_selection_string warnings.filterwarnings("ignore", category=dg.BetaWarning) class DbtAnalyticsConfig(dg.Config): full_refresh: bool = False class DbtAssetsFactory: """Factory for creating dbt assets with YAML-driven configuration.""" def __init__( self, dbt_project: DbtProject, dagster_dbt_translator: Optional[DagsterDbtTranslator] = None, default_backfill_policy: Optional[dg.BackfillPolicy] = None, default_automation_condition: Optional[AutomationCondition] = None, validate_missing_yaml: bool = True, ): self.dbt_project = dbt_project self.dagster_dbt_translator = dagster_dbt_translator self.default_backfill_policy = default_backfill_policy self.default_automation_condition = default_automation_condition self.validate_missing_yaml = validate_missing_yaml # Run validation if enabled if self.validate_missing_yaml: self._validate_sql_files_have_yaml_definitions() def _validate_sql_files_have_yaml_definitions(self) -> None: """Validate that all SQL model files have corresponding YAML definitions.""" models_dir = self.dbt_project.project_dir / "models" # Get all SQL model files sql_files = set() for sql_file in models_dir.rglob("*.sql"): # Extract model name from file path (remove .sql extension) model_name = sql_file.stem sql_files.add(model_name) # Get all models defined in YAML files yaml_models = set() for yaml_file in list(models_dir.rglob("*.yml")) + list(models_dir.rglob("*.yaml")): try: with open(yaml_file, 'r') as f: yaml_content = yaml.safe_load(f) if yaml_content and 'models' in yaml_content: for model in yaml_content['models']: model_name = model.get('name') if model_name: yaml_models.add(model_name) except Exception as e: print(f"Warning: Error reading {yaml_file}: {e}") continue # Find SQL files without YAML definitions missing_yaml = sql_files - yaml_models if missing_yaml: missing_models_list = sorted(list(missing_yaml)) raise ValueError( f"Found {len(missing_yaml)} SQL model(s) without corresponding YAML definitions. " f"All dbt models must be defined in YAML files for proper Dagster integration. " f"Missing YAML definitions for: {missing_models_list}. " f"Please create YAML model definitions for these files." ) def _read_model_metadata( self, tag_filter: Optional[str] = None, exclude_tags: Optional[List[str]] = None, by_partition: Optional[str] = None ) -> Dict[str, Dict[str, Any]]: """Read partition metadata from dbt manifest using DagsterDbtTranslator. Args: tag_filter: Optional tag to filter models (backward compatibility, deprecated) exclude_tags: Optional list of tags to exclude models by_partition: Optional partition type filter - 'hourly', 'daily', or 'unpartitioned' Returns: Dictionary mapping model names to their metadata dictionaries """ # Default exclude tags if exclude_tags is None: exclude_tags = ["exclude_from_dagster=true"] model_metadata = {} # Load the dbt manifest try: with open(self.dbt_project.manifest_path, 'r') as f: manifest = json.load(f) except json.JSONDecodeError as e: raise ValueError( f"Failed to parse dbt manifest at {self.dbt_project.manifest_path}: {e}. " f"The manifest file may be corrupted. Try running 'dbt parse' to regenerate it." ) from e except Exception as e: raise RuntimeError( f"Failed to read dbt manifest at {self.dbt_project.manifest_path}: {e}" ) from e # Get all model nodes from manifest nodes = manifest.get('nodes', {}) # Filter to only dbt models (not sources, exposures, etc.) model_nodes = { node_id: node_props for node_id, node_props in nodes.items() if node_props.get('resource_type') == 'model' } # Process each model node for node_id, node_props in model_nodes.items(): model_name = node_props.get('name') if not model_name: continue # Get tags from node props model_tags = node_props.get('tags', []) # Apply exclude tags if any(exclude_tag in model_tags for exclude_tag in exclude_tags): continue # Apply tag filter (backward compatibility, deprecated) if tag_filter and not any(tag_filter == tag for tag in model_tags): continue # Determine partition type using translator partition_type = None partitions_def_obj = None if self.dagster_dbt_translator: try: partitions_def_obj = self.dagster_dbt_translator.get_partitions_def(node_props) if partitions_def_obj: # Determine partition type from the PartitionsDefinition object if isinstance(partitions_def_obj, dg.HourlyPartitionsDefinition): partition_type = 'hourly' elif isinstance(partitions_def_obj, dg.DailyPartitionsDefinition): partition_type = 'daily' except Exception: # If translator fails or returns None, partition_type remains None (unpartitioned) pass # Filter by partition type if specified if by_partition == 'unpartitioned' and partition_type is not None: continue elif by_partition == 'daily' and partition_type != 'daily': continue elif by_partition == 'hourly' and partition_type != 'hourly': continue # Extract dagster metadata dagster_meta = node_props.get('meta', {}).get('dagster', {}) # Extract partitions_def dict for backward compatibility partitions_def_dict = dagster_meta.get('partitions_def', None) # Extract materialization type from config materialized = node_props.get('config', {}).get('materialized', 'view') # Store metadata in the same format as before model_metadata[model_name] = { 'partitions_def': partitions_def_dict, # Raw dict from YAML 'partition_start_date': dagster_meta.get('partition_start_date', DBT_MIN_PARTITION_DATE), 'end_offset': dagster_meta.get('end_offset', 0), 'max_partitions_per_backfill': dagster_meta.get('backfill_policy', {}).get('max_partitions_per_run'), 'dagster_meta': dagster_meta, # Store full dagster_meta for freshness policy 'materialized': materialized, # Store materialization type } return model_metadata def _read_exposure_metadata(self) -> Dict[str, Dict[str, Any]]: """Read exposure metadata from all exposure YAML files.""" models_dir = self.dbt_project.project_dir / "models" exposure_metadata = {} # Read through all exposure YAML files for yaml_file in list(models_dir.rglob("**/exposures/*.yml")) + list(models_dir.rglob("**/exposures/*.yaml")): try: with open(yaml_file, 'r') as f: yaml_content = yaml.safe_load(f) if not yaml_content or 'exposures' not in yaml_content: continue for exposure in yaml_content['exposures']: exposure_name = exposure.get('name') if not exposure_name: continue # Extract exposure metadata exposure_metadata[exposure_name] = { 'type': exposure.get('type', 'unknown'), 'description': exposure.get('description', ''), 'url': exposure.get('url', ''), 'owner': exposure.get('owner', {}), 'depends_on': exposure.get('depends_on', []), 'tags': exposure.get('tags', []), 'meta': exposure.get('meta', {}), } except Exception as e: print(f"Error reading exposure file {yaml_file}: {e}") continue return exposure_metadata def _read_snapshot_metadata(self, tag_filter: Optional[str] = None, exclude_tags: Optional[List[str]] = None) -> Dict[str, Dict[str, Any]]: """Read snapshot metadata from all snapshot YAML files.""" snapshots_dir = self.dbt_project.project_dir / "snapshots" snapshot_metadata = {} # Default exclude tags if exclude_tags is None: exclude_tags = ["exclude_from_dagster=true"] # Read through all snapshot YAML files for yaml_file in list(snapshots_dir.rglob("*.yml")) + list(snapshots_dir.rglob("*.yaml")): try: with open(yaml_file, 'r') as f: yaml_content = yaml.safe_load(f) if not yaml_content or 'snapshots' not in yaml_content: continue for snapshot in yaml_content['snapshots']: snapshot_name = snapshot.get('name') if not snapshot_name: continue # Check tags for filtering snapshot_config = snapshot.get('config', {}) snapshot_tags = snapshot_config.get('tags', []) if any(tag.startswith('partition_type=') for tag in snapshot_tags): raise ValueError( f"Snapshot '{snapshot_name}' has a partition_type tag. This is deprecated, please specify partition type in meta.dagster.partitions_def instead." ) # Apply tag filter if tag_filter and not any(tag_filter == tag for tag in snapshot_tags): continue # Apply exclude tags if any(exclude_tag in snapshot_tags for exclude_tag in exclude_tags): continue meta = snapshot.get('meta', {}) dagster_meta = meta.get('dagster', {}) # Extract snapshot metadata snapshot_metadata[snapshot_name] = { 'partitions_def': dagster_meta.get('partitions_def', None), } except Exception as e: print(f"Error reading snapshot file {yaml_file}: {e}") continue return snapshot_metadata def _create_backfill_policy(self, max_partitions_per_backfill: Optional[int] = None, partition_type: Optional[str] = None) -> Optional[dg.BackfillPolicy]: if max_partitions_per_backfill is not None: return dg.BackfillPolicy.multi_run(max_partitions_per_run=max_partitions_per_backfill) if partition_type == "daily": return dg.BackfillPolicy.multi_run(max_partitions_per_run=7*2) elif partition_type == "hourly": return dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2) return self.default_backfill_policy def _create_freshness_policy(self, partition_type: Optional[str], dagster_meta: Dict[str, Any]) -> Optional[FreshnessPolicy]: """ Create freshness policy based on partition type and metadata. Defaults (when no config provided): - Daily: warn if 27 hours, fail if 30 hours - Hourly: warn after 1.3 hours, fail after 1.7 hours Can be customized via meta.dagster.freshness_policy: For time_window type: - type: "time_window" (required) - warn_hours: float (optional, defaults based on partition_type) - fail_hours: float (optional, defaults based on partition_type) For cron type: - type: "cron" (required) - deadline_cron: string (required) - cron expression for deadline - lower_bound_delta_hours: int (required) - buffer time in hours - timezone: string (optional, defaults to "UTC") Example time_window: ```yaml meta: dagster: freshness_policy: type: time_window warn_hours: 27.0 fail_hours: 30.0 ``` Example cron: ```yaml meta: dagster: freshness_policy: type: cron deadline_cron: "0 0 * * *" # Daily at midnight lower_bound_delta_hours: 2 timezone: "UTC" ``` """ freshness_policy_config = dagster_meta.get('freshness_policy', None) # If no freshness policy config and no partition type, return None if not freshness_policy_config and not partition_type: return None # Determine default values based on partition type (for time_window only) if partition_type == "daily": default_warn_hours = 27.0 default_fail_hours = 30.0 elif partition_type == "hourly": default_warn_hours = 1.3 default_fail_hours = 1.7 else: # No partition type and no config, return None if not freshness_policy_config: return None # Partition type is None but config exists, require explicit values default_warn_hours = None default_fail_hours = None # Use config if provided, otherwise use defaults (time_window) if freshness_policy_config: policy_type = freshness_policy_config.get('type', 'time_window') if policy_type == 'time_window': warn_hours = freshness_policy_config.get('warn_hours', default_warn_hours) fail_hours = freshness_policy_config.get('fail_hours', default_fail_hours) if warn_hours is None or fail_hours is None: raise DagsterInvalidDefinitionError( f"warn_hours and fail_hours are required for time_window freshness policy when partition_type is not specified. " f"Got warn_hours: {warn_hours}, fail_hours: {fail_hours}" ) return FreshnessPolicy.time_window( warn_window=timedelta(hours=warn_hours), fail_window=timedelta(hours=fail_hours) ) elif policy_type == 'cron': deadline_cron = freshness_policy_config.get('deadline_cron', None) if not deadline_cron or not isinstance(deadline_cron, str): raise DagsterInvalidDefinitionError( f"deadline_cron (string) is required for cron freshness policy, but got deadline_cron: {deadline_cron}" ) if not is_valid_cron_string(deadline_cron): raise DagsterInvalidDefinitionError(f"Invalid cron string: {deadline_cron}") lower_bound_delta_hours = freshness_policy_config.get('lower_bound_delta_hours', None) if lower_bound_delta_hours is None or not isinstance(lower_bound_delta_hours, (int, float)): raise DagsterInvalidDefinitionError( f"lower_bound_delta_hours (int or float) is required for cron freshness policy, " f"but got lower_bound_delta_hours: {lower_bound_delta_hours}" ) lower_bound_delta = timedelta(hours=float(lower_bound_delta_hours)) timezone = freshness_policy_config.get('timezone', 'UTC') if not timezone or not isinstance(timezone, str): raise DagsterInvalidDefinitionError( f"timezone (string) is required for cron freshness policy, but got timezone: {timezone}" ) return FreshnessPolicy.cron( deadline_cron=deadline_cron, lower_bound_delta=lower_bound_delta, timezone=timezone ) else: raise DagsterInvalidDefinitionError( f"Unsupported freshness policy type: {policy_type}. Please use 'time_window' or 'cron'." ) else: # Use defaults (time_window) warn_hours = default_warn_hours fail_hours = default_fail_hours return FreshnessPolicy.time_window( warn_window=timedelta(hours=warn_hours), fail_window=timedelta(hours=fail_hours) ) def _create_partitions_def(self, partition_type: str, partition_start_date: str, end_offset: int = 0) -> dg.PartitionsDefinition: """Create appropriate PartitionsDefinition based on partition_type and start_date.""" if partition_type == "daily": return dg.DailyPartitionsDefinition(start_date=partition_start_date, end_offset=end_offset) elif partition_type == "hourly": return dg.HourlyPartitionsDefinition(start_date=partition_start_date, end_offset=end_offset) else: raise ValueError(f"Unsupported partition_type: {partition_type}") def _validate_manifest(self) -> None: """Validate that the dbt manifest exists and is accessible.""" if not self.dbt_project.manifest_path.exists(): raise FileNotFoundError( f"dbt manifest not found at {self.dbt_project.manifest_path}. " f"Please run 'dbt parse' or 'dbt compile' to generate the manifest." ) try: with open(self.dbt_project.manifest_path, 'r') as f: manifest = json.load(f) # Validate manifest has expected structure if not isinstance(manifest, dict): raise ValueError( f"dbt manifest at {self.dbt_project.manifest_path} is not a valid JSON object. " f"Please run 'dbt parse' to regenerate it." ) # Check for required keys (at least one of these should exist) if 'nodes' not in manifest and 'sources' not in manifest and 'exposures' not in manifest: raise ValueError( f"dbt manifest at {self.dbt_project.manifest_path} appears to be invalid or empty. " f"Expected at least one of: 'nodes', 'sources', 'exposures'. " f"Please run 'dbt parse' to regenerate it." ) except json.JSONDecodeError as e: raise ValueError( f"Failed to parse dbt manifest at {self.dbt_project.manifest_path}: {e}. " f"The manifest file may be corrupted. Try running 'dbt parse' to regenerate it." ) from e except Exception as e: if isinstance(e, (FileNotFoundError, ValueError)): raise raise RuntimeError( f"Failed to validate dbt manifest at {self.dbt_project.manifest_path}: {e}" ) from e def create_unpartitioned_assets( self, tag_filter: Optional[str] = None, exclude_tags: Optional[List[str]] = ["exclude_from_dagster=true"], ) -> List[dg.AssetsDefinition]: """Create unpartitioned dbt assets based on YAML metadata.""" # Validate manifest before creating assets self._validate_manifest() # Read model metadata filtered by unpartitioned models model_metadata = self._read_model_metadata(tag_filter=tag_filter, exclude_tags=exclude_tags, by_partition='unpartitioned') if not model_metadata: # Return empty list instead of raising an error when no models are found return [] # Create separate assets for each model with their individual configurations assets = [] for model_name, metadata in model_metadata.items(): @dbt_assets( name=model_name, manifest=self.dbt_project.manifest_path, dagster_dbt_translator=self.dagster_dbt_translator, select=model_name, exclude=get_dbt_tag_selection_string(exclude_tags, "ANY"), ) def _unpartitioned_dbt_model_asset(context: dg.AssetExecutionContext, dbt: DbtCliResource, config: DbtAnalyticsConfig): logger = dg.get_dagster_logger() dbt_build_args = ["build", "--debug"] if hasattr(config, 'full_refresh') and config.full_refresh: dbt_build_args.append("--full-refresh") logger.info(f"Building unpartitioned dbt model {model_name}") # Conditionally fetch row counts based on materialization type # Skip fetch_row_counts() for external tables to avoid counting entire external table dbt_result = dbt.cli(dbt_build_args, context=context).stream() if metadata.get('materialized') == 'external': yield from dbt_result.fetch_column_metadata() else: yield from dbt_result.fetch_row_counts().fetch_column_metadata() assets.append(_unpartitioned_dbt_model_asset) return assets def create_daily_partitioned_assets( self, tag_filter: Optional[str] = None, exclude_tags: Optional[List[str]] = ["exclude_from_dagster=true"], ) -> List[dg.AssetsDefinition]: """Create daily partitioned dbt assets based on YAML metadata.""" # Validate manifest before creating assets self._validate_manifest() # Read model metadata filtered by daily partitioned models model_metadata = self._read_model_metadata(tag_filter=tag_filter, exclude_tags=exclude_tags, by_partition='daily') if not model_metadata: # Return empty list instead of raising an error when no models are found return [] # Create separate assets for each model with their individual configurations assets = [] for model_name, metadata in model_metadata.items(): # TODO (@junseo): Remove this once we fully migrate partitions_def to DagsterDbtTranslator # If partitions_def is not set, use the old method of creating a daily partitions definition if not metadata['partitions_def']: partition_start_date = metadata['partition_start_date'] if isinstance(metadata['partition_start_date'], datetime) else datetime.strptime(metadata['partition_start_date'], "%Y-%m-%d") end_offset = metadata.get('end_offset', 0) # Create partitions definition for this specific model partitions_def = self._create_partitions_def( 'daily', partition_start_date, end_offset, ) else: partitions_def = None # Create backfill policy for this specific model backfill_policy = self._create_backfill_policy( metadata['max_partitions_per_backfill'], 'daily' ) @dbt_assets( name=model_name, manifest=self.dbt_project.manifest_path, dagster_dbt_translator=self.dagster_dbt_translator, select=model_name, exclude=get_dbt_tag_selection_string(exclude_tags, "ANY"), partitions_def=partitions_def, backfill_policy=backfill_policy, ) def _daily_partitioned_dbt_model_asset(context: dg.AssetExecutionContext, dbt: DbtCliResource, config: DbtAnalyticsConfig): logger = dg.get_dagster_logger() dbt_build_args = ["build", "--debug"] if hasattr(config, 'full_refresh') and config.full_refresh: dbt_build_args.append("--full-refresh") # Add partition time window to build vars logger.info(f"Building daily partitioned dbt model {model_name} for partitions {context.partition_key_range.start} to {context.partition_key_range.end}") partition_vars = { 'partition_start_date': context.partition_time_window.start.strftime("%Y-%m-%d"), 'partition_end_date': context.partition_time_window.end.strftime("%Y-%m-%d"), } logger.info(f"Partition vars: {partition_vars}") dbt_build_args.extend(["--vars", json.dumps(partition_vars)]) # Conditionally fetch row counts based on materialization type # Skip fetch_row_counts() for external tables to avoid counting entire external table # The dbt model is already filtered to the specific partition via partition vars dbt_result = dbt.cli(dbt_build_args, context=context).stream() if metadata.get('materialized') == 'external': yield from dbt_result.fetch_column_metadata() else: yield from dbt_result.fetch_row_counts().fetch_column_metadata() freshness_policy = self._create_freshness_policy('daily', metadata.get('dagster_meta', {})) if freshness_policy: _daily_partitioned_dbt_model_asset = _daily_partitioned_dbt_model_asset.map_asset_specs(lambda spec: apply_freshness_policy(spec, freshness_policy)) assets.append(_daily_partitioned_dbt_model_asset) return assets def create_hourly_partitioned_assets( self, tag_filter: Optional[str] = None, exclude_tags: Optional[List[str]] = None, ) -> List[dg.AssetsDefinition]: """Create hourly partitioned dbt assets based on YAML metadata.""" # Validate manifest before creating assets self._validate_manifest() # Read model metadata filtered by hourly partitioned models model_metadata = self._read_model_metadata(tag_filter=tag_filter, exclude_tags=exclude_tags, by_partition='hourly') if not model_metadata: # Return empty list instead of raising an error when no models are found return [] # Create separate assets for each model with their individual configurations assets = [] for model_name, metadata in model_metadata.items(): # TODO (@junseo): Remove this once we fully migrate partitions_def to DagsterDbtTranslator # If partitions_def is not set, use the old method of creating a hourly partitions definition if not metadata['partitions_def']: partition_start_date = metadata['partition_start_date'] if isinstance(metadata['partition_start_date'], datetime) else datetime.strptime(metadata['partition_start_date'], "%Y-%m-%d") end_offset = metadata.get('end_offset', 0) # Create partitions definition for this specific model partitions_def = self._create_partitions_def( 'hourly', partition_start_date, end_offset, ) else: partitions_def = None # Create backfill policy for this specific model backfill_policy = self._create_backfill_policy( metadata['max_partitions_per_backfill'], 'hourly' ) @dbt_assets( name=model_name, manifest=self.dbt_project.manifest_path, dagster_dbt_translator=self.dagster_dbt_translator, select=model_name, exclude=get_dbt_tag_selection_string(exclude_tags, "ANY"), partitions_def=partitions_def, backfill_policy=backfill_policy, ) def _hourly_partitioned_dbt_model_asset(context: dg.AssetExecutionContext, dbt: DbtCliResource, config: DbtAnalyticsConfig): logger = dg.get_dagster_logger() dbt_build_args = ["build", "--debug"] if hasattr(config, 'full_refresh') and config.full_refresh: dbt_build_args.append("--full-refresh") # Add partition time window to build vars logger.info(f"Building hourly partitioned dbt model {model_name} for partitions {context.partition_key_range.start} to {context.partition_key_range.end}") partition_vars = { 'partition_start_date': context.partition_time_window.start.strftime("%Y-%m-%d"), 'partition_end_date': context.partition_time_window.end.strftime("%Y-%m-%d"), 'partition_start_hour': context.partition_time_window.start.hour, 'partition_end_hour': context.partition_time_window.end.hour, } logger.info(f"Partition vars: {partition_vars}") dbt_build_args.extend(["--vars", json.dumps(partition_vars)]) # Conditionally fetch row counts based on materialization type # Skip fetch_row_counts() for external tables to avoid counting entire external table # The dbt model is already filtered to the specific partition via partition vars dbt_result = dbt.cli(dbt_build_args, context=context).stream() if metadata.get('materialized') == 'external': yield from dbt_result.fetch_column_metadata() else: yield from dbt_result.fetch_row_counts().fetch_column_metadata() freshness_policy = self._create_freshness_policy('hourly', metadata.get('dagster_meta', {})) if freshness_policy: _hourly_partitioned_dbt_model_asset = _hourly_partitioned_dbt_model_asset.map_asset_specs(lambda spec: apply_freshness_policy(spec, freshness_policy)) assets.append(_hourly_partitioned_dbt_model_asset) return assets def create_exposure_assets(self) -> List[AssetSpec]: """Create asset specs for dbt exposures using DagsterDbtTranslator to resolve dependencies.""" exposure_metadata = self._read_exposure_metadata() asset_specs = [] # Load the dbt manifest to resolve dependencies if not self.dbt_project.manifest_path.exists(): raise FileNotFoundError( f"dbt manifest not found at {self.dbt_project.manifest_path}. " f"Please run 'dbt parse' or 'dbt compile' to generate the manifest." ) try: with open(self.dbt_project.manifest_path, 'r') as f: manifest = json.load(f) except json.JSONDecodeError as e: raise ValueError( f"Failed to parse dbt manifest at {self.dbt_project.manifest_path}: {e}. " f"The manifest file may be corrupted. Try running 'dbt parse' to regenerate it." ) from e except Exception as e: raise RuntimeError( f"Failed to read dbt manifest at {self.dbt_project.manifest_path}: {e}" ) from e exposures = manifest.get('exposures', {}) nodes = manifest.get('nodes', {}) for exposure_name, metadata in exposure_metadata.items(): # Get exposure from manifest by finding matching exposure # Exposures in manifest have unique_id like "exposure.{project_name}.{exposure_name}" exposure_props = None for exp_unique_id, exp_props in exposures.items(): if exp_props.get('name') == exposure_name: exposure_props = exp_props break if not exposure_props: # If not found in manifest, log warning and continue with fallback print(f"Warning: Exposure {exposure_name} not found in manifest, using fallback dependency resolution") exposure_props = {} # Create metadata for the asset spec asset_metadata = { "exposure_type": metadata.get('type', 'unknown'), "owner_name": metadata.get('owner', {}).get('name', ''), "owner_email": metadata.get('owner', {}).get('email', ''), } # Add URL if available if metadata.get('url'): asset_metadata["url"] = metadata.get('url') # Resolve dependencies using DagsterDbtTranslator (same pattern as create_hourly_partitioned_assets) deps = [] if self.dagster_dbt_translator and exposure_props: # Get dependencies from manifest # depends_on can be a dict with 'nodes' key or a list depends_on_data = exposure_props.get('depends_on', {}) if isinstance(depends_on_data, dict): depends_on_nodes = depends_on_data.get('nodes', []) elif isinstance(depends_on_data, list): depends_on_nodes = depends_on_data else: depends_on_nodes = [] for node_unique_id in depends_on_nodes: # Get the node from manifest node_props = nodes.get(node_unique_id, {}) if not node_props: continue # Use translator to get asset key (same pattern as create_hourly_partitioned_assets) try: asset_key = self.dagster_dbt_translator.get_asset_key(node_props) if asset_key: deps.append(asset_key) except Exception as e: # Log warning but continue processing print(f"Warning: Could not resolve asset key for dependency {node_unique_id} in exposure {exposure_name}: {e}") continue else: # Fallback to old method if translator is not available for dep in metadata.get('depends_on', []): if isinstance(dep, dict) and 'name' in dep: deps.append(dg.AssetKey(dep['name'])) elif isinstance(dep, str): # Handle dbt ref() format - extract model name if dep.startswith('ref(') and dep.endswith(')'): # Extract model name from ref("model_name") model_name = dep[4:-1].strip('"\'') deps.append(dg.AssetKey(model_name)) else: deps.append(dg.AssetKey(dep)) # Create kinds based on dbt tags kinds = {"dbt",} # Add kinds from dbt tags dbt_tags = metadata.get('tags', []) if isinstance(dbt_tags, list): for tag in dbt_tags: if isinstance(tag, str) and "kind=" in tag.lower(): kinds.add(tag.split("=")[1]) # Create the asset spec asset_spec = AssetSpec( key=exposure_name, description=metadata.get('description', f"dbt exposure: {exposure_name}"), metadata=asset_metadata, deps=deps, group_name="dbt_exposures", kinds=kinds, ) asset_specs.append(asset_spec) return asset_specs def create_external_source_assets(self) -> List[AssetSpec]: """ Create AssetSpec for dbt sources that don't have meta.dagster.asset_key defined. These external assets are marked as ready by default so they don't block automation conditions. Follows the same pattern as create_exposure_assets(). Returns: List[AssetSpec]: List of asset specs for sources without explicit asset keys """ # Validate manifest before creating assets self._validate_manifest() asset_specs = [] # Load the dbt manifest try: with open(self.dbt_project.manifest_path, 'r') as f: manifest = json.load(f) except json.JSONDecodeError as e: raise ValueError( f"Failed to parse dbt manifest at {self.dbt_project.manifest_path}: {e}. " f"The manifest file may be corrupted. Try running 'dbt parse' to regenerate it." ) from e except Exception as e: raise RuntimeError( f"Failed to read dbt manifest at {self.dbt_project.manifest_path}: {e}" ) from e # Extract sources from manifest sources = manifest.get('sources', {}) for source_unique_id, source_props in sources.items(): # Get source metadata source_meta = source_props.get('meta', {}) dagster_meta = source_meta.get('dagster', {}) # Skip sources that already have an asset_key defined if dagster_meta.get('asset_key'): continue # Extract source information source_name = source_props.get('source_name') source_table_name = source_props.get('name') description = source_props.get('description', '') tags = source_props.get('tags', []) # Create asset key in format [source_name, table_name] asset_key = [source_name, source_table_name] # Create metadata for the asset spec asset_metadata = { "source_name": source_name, "source_table_name": source_table_name, "dbt_source_unique_id": source_unique_id, } # Add database and schema if available database = source_props.get('database') schema = source_props.get('schema') if database: asset_metadata["database"] = database if schema: asset_metadata["schema"] = schema # Extract partition_type from tags (similar to models) partition_type_from_tag = None if isinstance(tags, list): for tag in tags: if isinstance(tag, str) and tag.startswith("partition_type="): partition_type_from_tag = tag.split("=", 1)[1] break # Determine partition configuration from tags and meta # Default to daily if no partition_type is specified partition_type = partition_type_from_tag or "daily" # Store partition type in metadata for sensor use asset_metadata["partition_type"] = partition_type # Skip partition creation for non-partitioned sources partitions_def = None if partition_type == "none": # No partition configuration needed for non-partitioned sources asset_metadata["partition_date_column"] = None asset_metadata["partition_hour_column"] = None else: # Get partition start date and hour from meta start_date_str = dagster_meta.get('partition_start_date', DBT_MIN_PARTITION_DATE) start_hour = dagster_meta.get('partition_start_hour', 0) # Get custom partition column names (default to p_date and p_hour) partition_date_column = dagster_meta.get('partition_date_column', 'p_date') partition_hour_column = dagster_meta.get('partition_hour_column', 'p_hour') # Store partition configuration in metadata for sensor use if start_date_str: asset_metadata["partition_start_date"] = start_date_str asset_metadata["partition_start_hour"] = start_hour asset_metadata["partition_date_column"] = partition_date_column asset_metadata["partition_hour_column"] = partition_hour_column # Create partitions definition if we have start_date and partition_type is not "none" if start_date_str: start_date = pd.to_datetime(start_date_str) end_offset = dagster_meta.get('end_offset', -1 if partition_type == "hourly" else 0) if partition_type == "hourly": partitions_def = dg.HourlyPartitionsDefinition(start_date=start_date, end_offset=end_offset) elif partition_type == "daily": partitions_def = dg.DailyPartitionsDefinition(start_date=start_date, end_offset=end_offset) # Create kinds based on dbt tags kinds = {"dbt_source", "dbt"} # Add kinds from dbt tags if isinstance(tags, list): for tag in tags: if isinstance(tag, str) and "kind=" in tag.lower(): kinds.add(tag.split("=")[1]) # Determine group name group_name = dagster_meta.get('group') if not group_name: # Check source-level config for group source_config = source_props.get('config', {}) group_name = source_config.get('group', 'dbt_sources') # Create the asset spec (following exposure pattern) asset_spec = AssetSpec( key=asset_key, description=description or f"External dbt source: {source_name}.{source_table_name}", metadata=asset_metadata, deps=[], # External sources don't have dependencies group_name=group_name, kinds=kinds, partitions_def=partitions_def, ) asset_specs.append(asset_spec) return asset_specs def create_snapshot_assets( self, tag_filter: Optional[str] = None, exclude_tags: Optional[List[str]] = ["partition_type=hourly", "partition_type=daily", "exclude_from_dagster=true"], ) -> List[AssetSpec]: """ Create asset specs for dbt snapshots. Snapshots, unlike model assets, can be created without a SQL. They can simply have an entry in a YAML file in the snapshots directory. """ snapshot_metadata = self._read_snapshot_metadata(tag_filter, exclude_tags) if not snapshot_metadata: # Return empty list instead of raising an error when no snapshots are found return [] # Create separate assets for each snapshot with their individual configurations snapshot_assets = [] for snapshot_name, metadata in snapshot_metadata.items(): @dbt_assets( name=snapshot_name, manifest=self.dbt_project.manifest_path, dagster_dbt_translator=self.dagster_dbt_translator, select=snapshot_name, exclude=get_dbt_tag_selection_string(exclude_tags, "ANY"), ) def _snapshot_asset(context: dg.AssetExecutionContext, dbt: DbtCliResource, config: DbtAnalyticsConfig): logger = dg.get_dagster_logger() dbt_build_args = ["snapshot", "--debug"] logger.info(f"Building snapshot {snapshot_name}") yield from dbt.cli(dbt_build_args, context=context).stream().fetch_row_counts().fetch_column_metadata() snapshot_assets.append(_snapshot_asset) return snapshot_assets