Method create_data_source

sdk/python/feast/dbt/mapper.py:209–279

Create a Feast DataSource from a dbt model. Args: model: The DbtModel to create a DataSource from timestamp_field: Override the default timestamp field created_timestamp_column: Column for created timestamp (dedup) Returns: A

(
        self,
        model: DbtModel,
        timestamp_field: Optional[str] = None,
        created_timestamp_column: Optional[str] = None,
    )

        return ValueType.UNKNOWN

    def create_data_source(
        self,
        model: DbtModel,
        timestamp_field: Optional[str] = None,
        created_timestamp_column: Optional[str] = None,
    ) -> Any:
        """
        Create a Feast DataSource from a dbt model.

        Args:
            model: The DbtModel to create a DataSource from
            timestamp_field: Override the default timestamp field
            created_timestamp_column: Column for created timestamp (dedup)

        Returns:
            A Feast DataSource (BigQuerySource, SnowflakeSource, or FileSource)

        Raises:
            ValueError: If data_source_type is not supported
        """
        ts_field = timestamp_field or self.timestamp_field

        # Build tags from dbt metadata
        tags = {"dbt.model": model.name}
        for tag in model.tags:
            tags[f"dbt.tag.{tag}"] = "true"

        if self.data_source_type == "bigquery":
            from feast.infra.offline_stores.bigquery_source import BigQuerySource

            return BigQuerySource(
                name=f"{model.name}_source",
                table=model.full_table_name,
                timestamp_field=ts_field,
                created_timestamp_column=created_timestamp_column or "",
                description=model.description,
                tags=tags,
            )

        elif self.data_source_type == "snowflake":
            from feast.infra.offline_stores.snowflake_source import SnowflakeSource

            return SnowflakeSource(
                name=f"{model.name}_source",
                database=model.database,
                schema=model.schema,
                table=model.alias,
                timestamp_field=ts_field,
                created_timestamp_column=created_timestamp_column or "",
                description=model.description,
                tags=tags,
            )

        elif self.data_source_type == "file":
            from feast.infra.offline_stores.file_source import FileSource

            # For file sources, use the model name as a placeholder path
            return FileSource(