Function label_quality

sdk/python/feast/ui_server.py:276–418

Compute label quality metrics with conflict policy enforcement.

(request: LabelQualityRequest)

    @rest_app.post("/label-quality")
    def label_quality(request: LabelQualityRequest):
        """Compute label quality metrics with conflict policy enforcement."""
        try:
            from datetime import datetime, timezone

            from feast.labeling.conflict_policy import ConflictPolicy
            from feast.labeling.conflict_resolver import resolve_conflicts

            fv_name = request.feature_view
            fv: Any = None
            try:
                fv = store.registry.get_label_view(fv_name, store.project)
            except Exception:
                try:
                    fv = store.get_feature_view(fv_name)
                except Exception:
                    pass

            if fv is None:
                return Response(
                    content=json.dumps({"detail": f"Label view '{fv_name}' not found"}),
                    status_code=status.HTTP_404_NOT_FOUND,
                    media_type="application/json",
                )

            batch_source = getattr(fv, "batch_source", None)
            if batch_source is None:
                return Response(
                    content=json.dumps({"detail": f"No batch source for '{fv_name}'"}),
                    status_code=status.HTTP_400_BAD_REQUEST,
                    media_type="application/json",
                )

            feature_names = [f.name for f in fv.features]
            join_keys = (
                fv.join_keys
                if hasattr(fv, "join_keys")
                else (fv.entities if fv.entities else [])
            )

            provider = store._get_provider()
            timestamp_field = batch_source.timestamp_field
            labeler_field = getattr(fv, "labeler_field", "labeler")
            conflict_policy = getattr(
                fv, "conflict_policy", ConflictPolicy.LAST_WRITE_WINS
            )

            try:
                job = provider.offline_store.pull_all_from_table_or_query(
                    config=store.config,
                    data_source=batch_source,
                    join_key_columns=join_keys,
                    feature_name_columns=feature_names,
                    timestamp_field=timestamp_field,
                )
                raw_df = job.to_df()
            except Exception:
                return {