Method create_feature_view

sdk/python/feast/dbt/mapper.py:310–411

Create a Feast FeatureView from a dbt model. Args: model: The DbtModel to create a FeatureView from source: The DataSource for this FeatureView entity_columns: Entity column name(s) - single string or list of strings entities: Optiona

(
        self,
        model: DbtModel,
        source: Any,
        entity_columns: Union[str, List[str]],
        entities: Optional[Union[Entity, List[Entity]]] = None,
        timestamp_field: Optional[str] = None,
        ttl_days: Optional[int] = None,
        exclude_columns: Optional[List[str]] = None,
        online: bool = True,
    )

        )

    def create_feature_view(
        self,
        model: DbtModel,
        source: Any,
        entity_columns: Union[str, List[str]],
        entities: Optional[Union[Entity, List[Entity]]] = None,
        timestamp_field: Optional[str] = None,
        ttl_days: Optional[int] = None,
        exclude_columns: Optional[List[str]] = None,
        online: bool = True,
    ) -> FeatureView:
        """
        Create a Feast FeatureView from a dbt model.

        Args:
            model: The DbtModel to create a FeatureView from
            source: The DataSource for this FeatureView
            entity_columns: Entity column name(s) - single string or list of strings
            entities: Optional pre-created Entity or list of Entities
            timestamp_field: Override the default timestamp field
            ttl_days: Override the default TTL in days
            exclude_columns: Additional columns to exclude from features
            online: Whether to enable online serving

        Returns:
            A Feast FeatureView
        """
        # Normalize to lists
        entity_cols: List[str] = (
            [entity_columns]
            if isinstance(entity_columns, str)
            else list(entity_columns)
        )

        entity_objs: List[Entity] = []
        if entities is not None:
            entity_objs = [entities] if isinstance(entities, Entity) else list(entities)

        # Validate
        if not entity_cols:
            raise ValueError("At least one entity column must be specified")

        if entity_objs and len(entity_cols) != len(entity_objs):
            raise ValueError(
                f"Number of entity_columns ({len(entity_cols)}) must match "
                f"number of entities ({len(entity_objs)})"
            )

        ts_field = timestamp_field or self.timestamp_field
        ttl = timedelta(days=ttl_days if ttl_days is not None else self.ttl_days)

        # Columns to exclude from schema (timestamp + any explicitly excluded)
        # Note: entity columns should NOT be excluded - FeatureView.__init__
        # expects entity columns to be in the schema and will extract them
        excluded = {ts_field}
        if exclude_columns:
            excluded.update(exclude_columns)