import logging

import math

import time

from collections import defaultdict

from datetime import date, datetime, timedelta, timezone

from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple

from feast.feature_logging import LOG_TIMESTAMP_FIELD, FeatureServiceLoggingSource

from feast.infra.offline_stores.offline_store import OfflineStore

from feast.monitoring.dqm_job_manager import DQMJobManager

from feast.monitoring.metrics_calculator import MetricsCalculator

from feast.monitoring.monitoring_utils import build_view_aggregate

if TYPE_CHECKING:

from feast.feature_store import FeatureStore

logger = logging.getLogger(__name__)

VALID_GRANULARITIES = OfflineStore.MONITORING_VALID_GRANULARITIES

_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)

_FAR_FUTURE = datetime(2099, 12, 31, 23, 59, 59, tzinfo=timezone.utc)

GRANULARITY_WINDOWS = {

"daily": timedelta(days=1),

"weekly": timedelta(days=7),

"biweekly": timedelta(days=14),

"monthly": timedelta(days=30),

"quarterly": timedelta(days=90),

}

def _as_utc_datetime(val: Any) -> Optional[datetime]:

"""Parse a timestamp-like value to a timezone-aware UTC datetime."""

if val is None:

return None

if isinstance(val, datetime):

return val if val.tzinfo else val.replace(tzinfo=timezone.utc)

if isinstance(val, date):

return datetime.combine(val, datetime.min.time(), tzinfo=timezone.utc)

if isinstance(val, str):

parsed = datetime.fromisoformat(val.replace("Z", "+00:00"))

return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)

return None

def _newest_event_in_window(

max_ts: Optional[datetime],

start_dt: datetime,

end_dt: datetime,

) -> Optional[datetime]:

"""Newest source event that falls inside ``[start_dt, end_dt]``.

``max_ts`` is MAX(event_timestamp) over the whole source. If that

timestamp is after the window, the window end is used as an upper bound.

"""

aware_max = _as_utc_datetime(max_ts)

if aware_max is None or aware_max < start_dt:

return None

return min(aware_max, end_dt)

_FLOAT_FIELDS = frozenset(

{

"null_rate",

"mean",

"stddev",

"min_val",

"max_val",

"p50",

"p75",

"p90",

"p95",

"p99",

"avg_null_rate",

"max_null_rate",

}

)

def _sanitize_floats(row: Dict[str, Any]) -> Dict[str, Any]:

"""Replace NaN/Inf float values with None so JSON serialization succeeds."""

for key in _FLOAT_FIELDS:

val = row.get(key)

if isinstance(val, float) and (math.isnan(val) or math.isinf(val)):

row[key] = None

return row

class MonitoringService:

def __init__(self, store: "FeatureStore"):

self._store = store

self._job_manager: Optional[DQMJobManager] = None

self._calculator = MetricsCalculator()

self._monitoring_tables_ensured = False

self._offline_store_cache = None

def _get_offline_store(self):

if self._offline_store_cache is None:

self._offline_store_cache = self._store._get_provider().offline_store

return self._offline_store_cache

def _ensure_monitoring_tables(self):

if not self._monitoring_tables_ensured:

self._get_offline_store().ensure_monitoring_tables(self._store.config)

self._monitoring_tables_ensured = True

@property

def job_manager(self) -> DQMJobManager:

if self._job_manager is None:

self._job_manager = DQMJobManager(

self._get_offline_store(), self._store.config

)

self._job_manager.ensure_table()

return self._job_manager

# ------------------------------------------------------------------ #

# Auto-compute: detect dates, compute all granularities

# ------------------------------------------------------------------ #

def auto_compute(

self,

project: Optional[str] = None,

feature_view_name: Optional[str] = None,

allowed_feature_view_names: Optional[List[str]] = None,

) -> Dict[str, Any]:

"""Detect date ranges from source data and compute all granularities.

Also computes baseline for any features that don't have one yet

(idempotent -- features with existing baselines are skipped).

``allowed_feature_view_names`` optionally restricts a selector-free run to an

explicit set of views (e.g. the caller's authorized subset).

"""

start_time = time.time()

self._ensure_monitoring_tables()

if project is None:

project = self._store.config.project

feature_views = self._resolve_feature_views(

project, feature_view_name, allowed_feature_view_names

)

total_features = 0

total_views = 0

baseline_features = 0

granularities_computed = set()

for fv in feature_views:

try:

feature_fields = self._classify_fields(fv)

if not feature_fields:

continue

max_ts = self._get_max_timestamp(fv)

if max_ts is None:

logger.warning(

"No data found for feature view '%s', skipping", fv.name

)

continue

now = datetime.now(timezone.utc)

# Compute baseline for features that don't have one yet

fields_needing_baseline = self._get_features_without_baseline(

project, fv

)

if fields_needing_baseline:

bl_fields = self._classify_fields(

fv, fields=fields_needing_baseline

)

if bl_fields:

bl_metrics = self._compute_feature_metrics(

fv,

bl_fields,

_EPOCH,

_FAR_FUTURE,

)

self._save_computed_metrics(

project=project,

feature_view=fv,

metrics_list=bl_metrics,

metric_date=date.today(),

granularity="baseline",

set_baseline=True,

now=now,

max_event_timestamp=max_ts,

)

baseline_features += len(bl_metrics)

for granularity, window in GRANULARITY_WINDOWS.items():

window_start = max_ts - window

metrics_list = self._compute_feature_metrics(

fv,

feature_fields,

window_start,

max_ts,

)

self._save_computed_metrics(

project=project,

feature_view=fv,

metrics_list=metrics_list,

metric_date=window_start.date(),

granularity=granularity,

set_baseline=False,

now=now,

max_event_timestamp=max_ts,

)

self._compute_feature_service_metrics(

project=project,

granularity=granularity,

metric_dates=[window_start.date()],

set_baseline=False,

)

total_features += len(metrics_list)

granularities_computed.add(granularity)

total_views += 1

except Exception:

logger.exception(

"Failed to auto-compute metrics for feature view '%s'", fv.name

)

duration_ms = int((time.time() - start_time) * 1000)

return {

"status": "completed",

"computed_feature_views": total_views,

"computed_features": total_features,

"baseline_features": baseline_features,

"granularities": sorted(granularities_computed),

"duration_ms": duration_ms,

}

# ------------------------------------------------------------------ #

# Log source: compute metrics from feature serving logs

# ------------------------------------------------------------------ #

def compute_log_metrics(

self,

project: str,

feature_service_name: str,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

granularity: str = "daily",

set_baseline: bool = False,

) -> Dict[str, Any]:

"""Compute monitoring metrics from feature serving logs.

Requires the feature service to have a logging_config with a

LoggingDestination that can be converted to a DataSource.

"""

self._ensure_monitoring_tables()

if granularity not in VALID_GRANULARITIES:

raise ValueError(

f"Invalid granularity '{granularity}'. "

f"Must be one of {VALID_GRANULARITIES}"

)

start_time = time.time()

start_dt, end_dt = self._to_date_range(start_date, end_date)

fs = self._store.registry.get_feature_service(

name=feature_service_name, project=project

)

log_source = self._resolve_log_source(fs)

if log_source is None:

return {

"status": "skipped",

"reason": f"Feature service '{feature_service_name}' has no logging configured",

"duration_ms": int((time.time() - start_time) * 1000),

}

data_source, ts_field, feature_fields, log_col_map = log_source

metrics_list = self._compute_from_source(

data_source,

ts_field,

feature_fields,

start_dt,

end_dt,

)

now = datetime.now(timezone.utc)

metric_date = start_dt.date()

self._save_log_metrics(

project=project,

feature_service_name=feature_service_name,

log_col_map=log_col_map,

metrics_list=metrics_list,

metric_date=metric_date,

granularity=granularity,

set_baseline=set_baseline,

now=now,

max_event_timestamp=_newest_event_in_window(

self._get_max_timestamp_for_source(data_source, ts_field),

start_dt,

end_dt,

),

)

duration_ms = int((time.time() - start_time) * 1000)

return {

"status": "completed",

"data_source_type": "log",

"feature_service_name": feature_service_name,

"granularity": granularity,

"computed_features": len(metrics_list),

"metric_date": metric_date.isoformat(),

"duration_ms": duration_ms,

}

def auto_compute_log_metrics(

self,

project: Optional[str] = None,

feature_service_name: Optional[str] = None,

allowed_feature_service_names: Optional[List[str]] = None,

) -> Dict[str, Any]:

"""Auto-detect date ranges from log data and compute all granularities.

``allowed_feature_service_names`` optionally restricts a selector-free run to an

explicit set of services (e.g. the caller's authorized subset).

"""

start_time = time.time()

self._ensure_monitoring_tables()

if project is None:

project = self._store.config.project

if feature_service_name:

services = [

self._store.registry.get_feature_service(

name=feature_service_name, project=project

)

]

elif allowed_feature_service_names is not None:

services = [

self._store.registry.get_feature_service(name=name, project=project)

for name in allowed_feature_service_names

]

else:

services = self._store.registry.list_feature_services(project=project)

total_features = 0

total_services = 0

granularities_computed: set = set()

for fs in services:

try:

log_source = self._resolve_log_source(fs)

if log_source is None:

continue

data_source, ts_field, feature_fields, log_col_map = log_source

max_ts = self._get_max_timestamp_for_source(data_source, ts_field)

if max_ts is None:

logger.warning(

"No log data found for feature service '%s', skipping",

fs.name,

)

continue

now = datetime.now(timezone.utc)

for gran, window in GRANULARITY_WINDOWS.items():

window_start = max_ts - window

metrics_list = self._compute_from_source(

data_source,

ts_field,

feature_fields,

window_start,

max_ts,

)

self._save_log_metrics(

project=project,

feature_service_name=fs.name,

log_col_map=log_col_map,

metrics_list=metrics_list,

metric_date=window_start.date(),

granularity=gran,

set_baseline=False,

now=now,

max_event_timestamp=max_ts,

)

total_features += len(metrics_list)

granularities_computed.add(gran)

total_services += 1

except Exception:

logger.exception(

"Failed to auto-compute log metrics for feature service '%s'",

fs.name,

)

duration_ms = int((time.time() - start_time) * 1000)

return {

"status": "completed",

"data_source_type": "log",

"computed_feature_services": total_services,

"computed_features": total_features,

"granularities": sorted(granularities_computed),

"duration_ms": duration_ms,

}

# ------------------------------------------------------------------ #

# Baseline: compute from all available source data

# ------------------------------------------------------------------ #

def compute_baseline(

self,

project: Optional[str] = None,

feature_view_name: Optional[str] = None,

feature_names: Optional[List[str]] = None,

allowed_feature_view_names: Optional[List[str]] = None,

) -> Dict[str, Any]:

"""Compute baseline metrics from all available source data.

Idempotent: only features without existing baselines are computed.

``allowed_feature_view_names`` optionally restricts a selector-free run to an

explicit set of views (e.g. the caller's authorized subset).

"""

start_time = time.time()

self._ensure_monitoring_tables()

if project is None:

project = self._store.config.project

feature_views = self._resolve_feature_views(

project, feature_view_name, allowed_feature_view_names

)

total_features = 0

total_views = 0

for fv in feature_views:

try:

fields_needing_baseline = self._get_features_without_baseline(

project, fv, feature_names

)

if not fields_needing_baseline:

logger.info(

"All features in '%s' already have baselines, skipping",

fv.name,

)

continue

feature_fields = self._classify_fields(

fv, fields=fields_needing_baseline

)

if not feature_fields:

continue

metrics_list = self._compute_feature_metrics(

fv,

feature_fields,

_EPOCH,

_FAR_FUTURE,

)

now = datetime.now(timezone.utc)

self._save_computed_metrics(

project=project,

feature_view=fv,

metrics_list=metrics_list,

metric_date=date.today(),

granularity="baseline",

set_baseline=True,

now=now,

max_event_timestamp=self._get_max_timestamp(fv),

)

total_features += len(metrics_list)

total_views += 1

except Exception:

logger.exception(

"Failed to compute baseline for feature view '%s'", fv.name

)

duration_ms = int((time.time() - start_time) * 1000)

return {

"status": "completed",

"computed_features": total_features,

"computed_feature_views": total_views,

"is_baseline": True,

"duration_ms": duration_ms,

}

# ------------------------------------------------------------------ #

# Compute: explicit dates + granularity (stored)

# ------------------------------------------------------------------ #

def compute_metrics(

self,

project: str,

feature_view_name: Optional[str] = None,

feature_names: Optional[List[str]] = None,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

granularity: str = "daily",

set_baseline: bool = False,

allowed_feature_view_names: Optional[List[str]] = None,

) -> Dict[str, Any]:

self._ensure_monitoring_tables()

if granularity not in VALID_GRANULARITIES:

raise ValueError(

f"Invalid granularity '{granularity}'. "

f"Must be one of {VALID_GRANULARITIES}"

)

start_time = time.time()

start_dt, end_dt = self._to_date_range(start_date, end_date)

feature_views = self._resolve_feature_views(

project, feature_view_name, allowed_feature_view_names

)

total_features = 0

total_views = 0

computed_dates: set = set()

for fv in feature_views:

try:

max_ts = self._get_max_timestamp(fv)

fv_metrics = self._compute_for_feature_view(

project=project,

feature_view=fv,

feature_names=feature_names,

start_dt=start_dt,

end_dt=end_dt,

granularity=granularity,

set_baseline=set_baseline,

max_event_timestamp=max_ts,

)

total_features += fv_metrics["feature_count"]

total_views += 1

computed_dates.update(fv_metrics["dates"])

except Exception:

logger.exception(

"Failed to compute metrics for feature view '%s'", fv.name

)

total_services = self._compute_feature_service_metrics(

project=project,

granularity=granularity,

metric_dates=list(computed_dates),

set_baseline=set_baseline,

)

duration_ms = int((time.time() - start_time) * 1000)

return {

"status": "completed",

"granularity": granularity,

"computed_features": total_features,

"computed_feature_views": total_views,

"computed_feature_services": total_services,

"metric_dates": sorted(d.isoformat() for d in computed_dates),

"duration_ms": duration_ms,

}

# ------------------------------------------------------------------ #

# Transient compute (not stored)

# ------------------------------------------------------------------ #

def compute_transient(

self,

project: str,

feature_view_name: str,

feature_names: Optional[List[str]] = None,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

) -> Dict[str, Any]:

"""Compute metrics on-the-fly for an arbitrary date range without persisting."""

start_time = time.time()

start_dt, end_dt = self._to_date_range(start_date, end_date)

effective_start = start_date or (date.today() - timedelta(days=1))

effective_end = end_date or date.today()

fv = self._store.registry.get_feature_view(

name=feature_view_name, project=project

)

feature_fields = self._classify_fields(fv, feature_names=feature_names)

if not feature_fields:

return {

"status": "completed",

"feature_view_name": feature_view_name,

"start_date": effective_start.isoformat(),

"end_date": effective_end.isoformat(),

"metrics": [],

"duration_ms": int((time.time() - start_time) * 1000),

}

metrics_list = self._compute_feature_metrics(

fv,

feature_fields,

start_dt,

end_dt,

)

for m in metrics_list:

m["feature_view_name"] = feature_view_name

m["start_date"] = effective_start.isoformat()

m["end_date"] = effective_end.isoformat()

return {

"status": "completed",

"feature_view_name": feature_view_name,

"start_date": effective_start.isoformat(),

"end_date": effective_end.isoformat(),

"metrics": metrics_list,

"duration_ms": int((time.time() - start_time) * 1000),

}

# ------------------------------------------------------------------ #

# DQM Job helpers

# ------------------------------------------------------------------ #

def submit_job(

self,

project: str,

job_type: str,

feature_view_name: Optional[str] = None,

parameters: Optional[Dict[str, Any]] = None,

) -> str:

return self.job_manager.submit(

project=project,

job_type=job_type,

feature_view_name=feature_view_name,

parameters=parameters,

)

def get_job(

self, job_id: str, project: Optional[str] = None

) -> Optional[Dict[str, Any]]:

return self.job_manager.get_job(job_id, project=project)

def execute_job(self, job_id: str) -> Dict[str, Any]:

return self.job_manager.execute_job(job_id, self)

# ------------------------------------------------------------------ #

# Read helpers (delegate to offline store)

# ------------------------------------------------------------------ #

def _query(

self,

metric_type: str,

project: str,

filters=None,

start_date=None,

end_date=None,

):

self._ensure_monitoring_tables()

rows = self._get_offline_store().query_monitoring_metrics(

config=self._store.config,

project=project,

metric_type=metric_type,

filters=filters,

start_date=start_date,

end_date=end_date,

)

return [_sanitize_floats(r) for r in rows]

def get_feature_metrics(

self,

project: str,

feature_service_name: Optional[str] = None,

feature_view_name: Optional[str] = None,

feature_name: Optional[str] = None,

granularity: Optional[str] = None,

data_source_type: Optional[str] = None,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

) -> List[Dict[str, Any]]:

filters = {

"feature_view_name": feature_view_name,

"feature_name": feature_name,

"granularity": granularity,

"data_source_type": data_source_type,

}

if feature_service_name:

return self._get_metrics_by_service(

project,

feature_service_name,

lambda fv_name: self._query(

"feature",

project,

{**filters, "feature_view_name": fv_name},

start_date,

end_date,

),

)

return self._query("feature", project, filters, start_date, end_date)

def get_feature_view_metrics(

self,

project: str,

feature_service_name: Optional[str] = None,

feature_view_name: Optional[str] = None,

granularity: Optional[str] = None,

data_source_type: Optional[str] = None,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

) -> List[Dict[str, Any]]:

filters = {

"feature_view_name": feature_view_name,

"granularity": granularity,

"data_source_type": data_source_type,

}

if feature_service_name:

return self._get_metrics_by_service(

project,

feature_service_name,

lambda fv_name: self._query(

"feature_view",

project,

{**filters, "feature_view_name": fv_name},

start_date,

end_date,

),

)

return self._query("feature_view", project, filters, start_date, end_date)

def get_feature_service_metrics(

self,

project: str,

feature_service_name: Optional[str] = None,

granularity: Optional[str] = None,

data_source_type: Optional[str] = None,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

) -> List[Dict[str, Any]]:

filters = {

"feature_service_name": feature_service_name,

"granularity": granularity,

"data_source_type": data_source_type,

}

return self._query("feature_service", project, filters, start_date, end_date)

def get_baseline(

self,

project: str,

feature_view_name: Optional[str] = None,

feature_name: Optional[str] = None,

data_source_type: Optional[str] = None,

) -> List[Dict[str, Any]]:

filters = {

"feature_view_name": feature_view_name,

"feature_name": feature_name,

"data_source_type": data_source_type,

"is_baseline": True,

}

return self._query("feature", project, filters)

def get_timeseries(

self,

project: str,

feature_view_name: Optional[str] = None,

feature_name: Optional[str] = None,

feature_service_name: Optional[str] = None,

granularity: Optional[str] = None,

data_source_type: Optional[str] = None,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

) -> List[Dict[str, Any]]:

return self.get_feature_metrics(

project=project,

feature_service_name=feature_service_name,

feature_view_name=feature_view_name,

feature_name=feature_name,

granularity=granularity,

data_source_type=data_source_type,

start_date=start_date,

end_date=end_date,

)

# ------------------------------------------------------------------ #

# Auto-baseline trigger for feast apply

# ------------------------------------------------------------------ #

def submit_baseline_for_new_features(

self,

project: str,

feature_views: Optional[List] = None,

) -> List[str]:

"""Submit baseline DQM jobs for feature views with new features.

Called from feast apply. Returns list of submitted job IDs.

Idempotent — only features without existing baselines are included.

"""

if project is None:

project = self._store.config.project

if feature_views is None:

feature_views = self._store.registry.list_feature_views(project=project)

job_ids = []

for fv in feature_views:

new_features = self._get_features_without_baseline(project, fv)

if not new_features:

continue

feature_names = [f.name for f in new_features]

job_id = self.job_manager.submit(

project=project,

job_type="baseline",

feature_view_name=fv.name,

parameters={"feature_names": feature_names},

)

job_ids.append(job_id)

logger.info(

"Queued baseline computation for '%s' features %s (job: %s)",

fv.name,

feature_names,

job_id,

)

return job_ids

# ------------------------------------------------------------------ #

# Private: compute engine dispatch (SQL push-down → Python fallback)

# ------------------------------------------------------------------ #

def _compute_feature_metrics(

self,

feature_view,

feature_fields: List[Tuple[str, str]],

start_dt: datetime,

end_dt: datetime,

) -> List[Dict[str, Any]]:

"""Compute metrics from a feature view's batch source."""

return self._compute_from_source(

feature_view.batch_source,

feature_view.batch_source.timestamp_field,

feature_fields,

start_dt,

end_dt,

)

def _get_max_timestamp(self, feature_view) -> Optional[datetime]:

"""Query the batch source for MAX(event_timestamp)."""

return self._get_max_timestamp_for_source(

feature_view.batch_source,

feature_view.batch_source.timestamp_field,

)

# ------------------------------------------------------------------ #

# Private: shared helpers (DRY)

# ------------------------------------------------------------------ #

@staticmethod

def _to_date_range(

start_date: Optional[date], end_date: Optional[date]

) -> Tuple[datetime, datetime]:

today = date.today()

if end_date is None:

end_date = today

if start_date is None:

start_date = end_date - timedelta(days=1)

start_dt = datetime(

start_date.year, start_date.month, start_date.day, tzinfo=timezone.utc

)

end_dt = datetime(

end_date.year, end_date.month, end_date.day, 23, 59, 59, tzinfo=timezone.utc

)

return start_dt, end_dt

@staticmethod

def _classify_fields(

feature_view,

feature_names=None,

fields=None,

) -> List[Tuple[str, str]]:

"""Extract and classify features as numeric/categorical.

Args:

feature_view: FeatureView to extract fields from (used if fields is None).

feature_names: Optional filter list of feature names.

fields: Optional pre-selected Field objects (e.g., from idempotency check).

"""

if fields is None:

fields = feature_view.features

if feature_names:

fields = [f for f in fields if f.name in feature_names]

result = []

for field in fields:

ftype = MetricsCalculator.classify_feature(field.dtype)

if ftype is None:

logger.warning(

"Unsupported dtype '%s' for feature '%s', skipping",

field.dtype,

field.name,

)

continue

result.append((field.name, ftype))

return result

def _save_computed_metrics(

self,

project: str,

feature_view,

metrics_list: List[Dict[str, Any]],

metric_date: date,

granularity: str,

set_baseline: bool,

now: datetime,

max_event_timestamp: Optional[datetime] = None,

) -> None:

if not metrics_list:

return

offline_store = self._get_offline_store()

config = self._store.config

if set_baseline:

offline_store.clear_monitoring_baseline(

config=config,

project=project,

feature_view_name=feature_view.name,

)

for m in metrics_list:

m["project_id"] = project

m["feature_view_name"] = feature_view.name

m["metric_date"] = metric_date

m["granularity"] = granularity

m["data_source_type"] = "batch"

m["computed_at"] = now

m["max_event_timestamp"] = max_event_timestamp

m["is_baseline"] = set_baseline

offline_store.save_monitoring_metrics(config, "feature", metrics_list)

view_metric = {

"project_id": project,

"feature_view_name": feature_view.name,

"metric_date": metric_date,

"granularity": granularity,

"data_source_type": "batch",

"computed_at": now,

"max_event_timestamp": max_event_timestamp,

"is_baseline": set_baseline,

**build_view_aggregate(metrics_list),

}

offline_store.save_monitoring_metrics(config, "feature_view", [view_metric])

def _resolve_join_key_columns(self, feature_view) -> List[str]:

config = self._store.config

return (

[

entity.name

for entity in self._store.registry.list_entities(project=config.project)

if entity.name in (feature_view.entities or [])

]

or feature_view.entities

or []

)

def _get_metrics_by_service(

self, project: str, feature_service_name: str, query_fn

):

fs = self._store.registry.get_feature_service(

name=feature_service_name, project=project

)

fv_names = [proj.name for proj in fs.feature_view_projections]

results = []

for fv_name in fv_names:

results.extend(query_fn(fv_name))

return results

def _resolve_feature_views(

self,

project: str,

feature_view_name: Optional[str],

allowed_feature_view_names: Optional[List[str]] = None,

):

if feature_view_name:

fv = self._store.registry.get_feature_view(

name=feature_view_name, project=project

)

return [fv]

if allowed_feature_view_names is not None:

# Explicit allow-list (e.g. the caller's authorized subset for a

# selector-free request). An empty list resolves to no views.

return [

self._store.registry.get_feature_view(name=name, project=project)

for name in allowed_feature_view_names

]

return self._store.registry.list_feature_views(project=project)

def _get_features_without_baseline(self, project, feature_view, feature_names=None):

existing = self.get_baseline(

project=project,

feature_view_name=feature_view.name,

)

existing_names = {m["feature_name"] for m in existing}

fields = feature_view.features

if feature_names:

fields = [f for f in fields if f.name in feature_names]