feat: Enable static artifacts for feature server that can be used in …

…Feature Transformations Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
feast-dev · franciscojavierarceo · Dec 23, 2025 · Dec 19, 2025 · Dec 19, 2025 · Dec 19, 2025
commit 358f3b5aaa9147bee02a16ec124f7d83701ae9c2
@@ -181,13 +181,15 @@ async def load_static_artifacts(app: FastAPI, store):
 
         if artifacts_file.exists():
             # Load and execute custom static artifacts loading
-            spec = importlib.util.spec_from_file_location("static_artifacts", artifacts_file)
+            spec = importlib.util.spec_from_file_location(
+                "static_artifacts", artifacts_file
+            )
             if spec and spec.loader:
                 artifacts_module = importlib.util.module_from_spec(spec)
                 spec.loader.exec_module(artifacts_module)
 
                 # Look for load_artifacts function
-                if hasattr(artifacts_module, 'load_artifacts'):
+                if hasattr(artifacts_module, "load_artifacts"):
                     load_func = artifacts_module.load_artifacts
                     if inspect.iscoroutinefunction(load_func):
                         await load_func(app)
@@ -259,7 +261,6 @@ def async_refresh():
             active_timer = threading.Timer(registry_ttl_sec, async_refresh)
             active_timer.start()
 
-
     @asynccontextmanager
     async def lifespan(app: FastAPI):
         # Load static artifacts before initializing store

@@ -27,7 +27,7 @@
 
 try:
     # Import static artifacts helpers (available when feature server loads artifacts)
-    from static_artifacts import get_sentiment_model, get_lookup_tables
+    from static_artifacts import get_lookup_tables, get_sentiment_model
 except ImportError:
     # Fallback for when static_artifacts.py is not available
     get_sentiment_model = None
@@ -165,7 +165,6 @@ def sentiment_prediction(inputs: pd.DataFrame) -> pd.DataFrame:
         import numpy as np
     except ImportError:
         # Fallback to dummy predictions if numpy isn't available
-        import array as np_fallback
 
         df = pd.DataFrame()
         df["predicted_sentiment"] = ["neutral"] * len(inputs)
@@ -184,11 +183,10 @@ def sentiment_prediction(inputs: pd.DataFrame) -> pd.DataFrame:
     lookup_tables = _lookup_tables
 
     # Use lookup table for label mapping (from static artifacts)
-    label_map = lookup_tables.get("sentiment_labels", {
-        "LABEL_0": "negative",
-        "LABEL_1": "neutral",
-        "LABEL_2": "positive"
-    })
+    label_map = lookup_tables.get(
+        "sentiment_labels",
+        {"LABEL_0": "negative", "LABEL_1": "neutral", "LABEL_2": "positive"},
+    )
 
     results = []
 
@@ -206,7 +204,9 @@ def sentiment_prediction(inputs: pd.DataFrame) -> pd.DataFrame:
 
                 # Get best prediction
                 best_pred = max(predictions, key=lambda x: x["score"])
-                predicted_sentiment = label_map.get(best_pred["label"], best_pred["label"])
+                predicted_sentiment = label_map.get(
+                    best_pred["label"], best_pred["label"]
+                )
                 confidence = best_pred["score"]
             else:
                 # Fallback when model is not available
@@ -217,25 +217,29 @@ def sentiment_prediction(inputs: pd.DataFrame) -> pd.DataFrame:
             # Generate dummy embeddings (in production, use pre-loaded embeddings)
             embedding = np.random.rand(384).tolist()
 
-            results.append({
-                "predicted_sentiment": predicted_sentiment,
-                "sentiment_confidence": np.float32(confidence),
-                "positive_prob": np.float32(scores.get("positive", 0.0)),
-                "negative_prob": np.float32(scores.get("negative", 0.0)),
-                "neutral_prob": np.float32(scores.get("neutral", 0.0)),
-                "text_embedding": [np.float32(x) for x in embedding],
-            })
+            results.append(
+                {
+                    "predicted_sentiment": predicted_sentiment,
+                    "sentiment_confidence": np.float32(confidence),
+                    "positive_prob": np.float32(scores.get("positive", 0.0)),
+                    "negative_prob": np.float32(scores.get("negative", 0.0)),
+                    "neutral_prob": np.float32(scores.get("neutral", 0.0)),
+                    "text_embedding": [np.float32(x) for x in embedding],
+                }
+            )
 
         except Exception:
             # Fallback for individual text processing errors
-            results.append({
-                "predicted_sentiment": "neutral",
-                "sentiment_confidence": np.float32(0.5),
-                "positive_prob": np.float32(0.33),
-                "negative_prob": np.float32(0.33),
-                "neutral_prob": np.float32(0.34),
-                "text_embedding": [np.float32(0.0)] * 384,
-            })
+            results.append(
+                {
+                    "predicted_sentiment": "neutral",
+                    "sentiment_confidence": np.float32(0.5),
+                    "positive_prob": np.float32(0.33),
+                    "negative_prob": np.float32(0.33),
+                    "neutral_prob": np.float32(0.34),
+                    "text_embedding": [np.float32(0.0)] * 384,
+                }
+            )
 
     return pd.DataFrame(results)
 

@@ -14,7 +14,6 @@
 use dedicated model serving solutions like vLLM, TensorRT-LLM, or TGI.
 """
 
-import asyncio
 from pathlib import Path
 from typing import Any, Dict, Optional
 
@@ -38,7 +37,9 @@ def load_sentiment_model():
         logger.info("✅ Sentiment analysis model loaded successfully")
         return model
     except ImportError:
-        logger.warning("⚠️ Transformers not available, sentiment model will use fallback")
+        logger.warning(
+            "⚠️ Transformers not available, sentiment model will use fallback"
+        )
         return None
     except Exception as e:
         logger.warning(f"⚠️ Failed to load sentiment model: {e}")
@@ -49,7 +50,11 @@ def load_lookup_tables() -> Dict[str, Any]:
     """Load static lookup tables for feature engineering."""
     # Example: Load static mappings that are expensive to compute at request time
     return {
-        "sentiment_labels": {"LABEL_0": "negative", "LABEL_1": "neutral", "LABEL_2": "positive"},
+        "sentiment_labels": {
+            "LABEL_0": "negative",
+            "LABEL_1": "neutral",
+            "LABEL_2": "positive",
+        },
         "emoji_sentiment": {"😊": "positive", "😞": "negative", "😐": "neutral"},
         "domain_categories": {"twitter.com": "social", "news.com": "news"},
     }
@@ -63,6 +68,7 @@ def load_user_embeddings() -> Optional[Dict[str, Any]]:
     if embeddings_file.exists():
         try:
             import numpy as np
+
             embeddings = np.load(embeddings_file)
             logger.info(f"✅ Loaded user embeddings: {embeddings.shape}")
             return {"embeddings": embeddings}
@@ -93,6 +99,7 @@ def load_artifacts(app: FastAPI):
     # Also set global references for easier access from on-demand feature views
     try:
         import example_repo
+
         example_repo._sentiment_model = app.state.sentiment_model
         example_repo._lookup_tables = app.state.lookup_tables
         logger.info("✅ Global artifact references updated")
@@ -129,4 +136,4 @@ def get_lookup_tables(app_state: Any) -> Dict[str, Any]:
 
 def get_user_embeddings(app_state: Any):
     """Get the pre-loaded user embeddings."""
-    return get_static_artifact(app_state, "user_embeddings")
+    return get_static_artifact(app_state, "user_embeddings")
@@ -247,7 +247,9 @@ def load_artifacts(app: FastAPI):
 def test_load_static_artifacts_success(mock_store_with_static_artifacts):
     """Test successful loading of static artifacts during server startup."""
     import asyncio
+
     from fastapi import FastAPI
+
     from feast.feature_server import load_static_artifacts
 
     app = FastAPI()
@@ -256,16 +258,18 @@ def test_load_static_artifacts_success(mock_store_with_static_artifacts):
     asyncio.run(load_static_artifacts(app, mock_store_with_static_artifacts))
 
     # Verify artifacts were loaded into app.state
-    assert hasattr(app.state, 'test_model')
-    assert hasattr(app.state, 'test_lookup_tables')
+    assert hasattr(app.state, "test_model")
+    assert hasattr(app.state, "test_lookup_tables")
     assert app.state.test_model == "test_model_loaded"
     assert app.state.test_lookup_tables == {"test_label": "test_value"}
 
 
 def test_load_static_artifacts_no_file(tmp_path):
     """Test graceful handling when static_artifacts.py doesn't exist."""
     import asyncio
+
     from fastapi import FastAPI
+
     from feast.feature_server import load_static_artifacts
 
     app = FastAPI()
@@ -276,14 +280,16 @@ def test_load_static_artifacts_no_file(tmp_path):
     asyncio.run(load_static_artifacts(app, mock_store))
 
     # Should not have added test artifacts
-    assert not hasattr(app.state, 'test_model')
-    assert not hasattr(app.state, 'test_lookup_tables')
+    assert not hasattr(app.state, "test_model")
+    assert not hasattr(app.state, "test_lookup_tables")
 
 
 def test_load_static_artifacts_invalid_file(tmp_path):
     """Test graceful handling when static_artifacts.py has errors."""
     import asyncio
+
     from fastapi import FastAPI
+
     from feast.feature_server import load_static_artifacts
 
     # Create invalid static_artifacts.py
@@ -298,13 +304,15 @@ def test_load_static_artifacts_invalid_file(tmp_path):
     asyncio.run(load_static_artifacts(app, mock_store))
 
     # Should not have artifacts due to error
-    assert not hasattr(app.state, 'test_model')
+    assert not hasattr(app.state, "test_model")
 
 
 def test_load_static_artifacts_no_load_function(tmp_path):
     """Test handling when static_artifacts.py has no load_artifacts function."""
     import asyncio
+
     from fastapi import FastAPI
+
     from feast.feature_server import load_static_artifacts
 
     # Create static_artifacts.py without load_artifacts function
@@ -319,13 +327,12 @@ def test_load_static_artifacts_no_load_function(tmp_path):
     asyncio.run(load_static_artifacts(app, mock_store))
 
     # Should not have artifacts since no load_artifacts function
-    assert not hasattr(app.state, 'test_model')
+    assert not hasattr(app.state, "test_model")
 
 
 def test_static_artifacts_persist_across_requests(mock_store_with_static_artifacts):
     """Test that static artifacts persist across multiple requests."""
     from feast.feature_server import get_app
-    from fastapi.testclient import TestClient
 
     # Create app with static artifacts
     app = get_app(mock_store_with_static_artifacts)
@@ -334,8 +341,6 @@ def test_static_artifacts_persist_across_requests(mock_store_with_static_artifac
     app.state.test_model = "persistent_model"
     app.state.test_lookup_tables = {"persistent": "data"}
 
-    client = TestClient(app)
-
     # Artifacts should be available and persistent
     assert app.state.test_model == "persistent_model"
     assert app.state.test_lookup_tables["persistent"] == "data"
@@ -348,7 +353,9 @@ def test_static_artifacts_persist_across_requests(mock_store_with_static_artifac
 def test_pytorch_nlp_template_artifacts_pattern(tmp_path):
     """Test the specific PyTorch NLP template static artifacts pattern."""
     import asyncio
+
     from fastapi import FastAPI
+
     from feast.feature_server import load_static_artifacts
 
     # Create PyTorch NLP template-style static_artifacts.py
@@ -389,8 +396,8 @@ def load_artifacts(app: FastAPI):
     asyncio.run(load_static_artifacts(app, mock_store))
 
     # Verify PyTorch NLP template artifacts
-    assert hasattr(app.state, 'sentiment_model')
-    assert hasattr(app.state, 'lookup_tables')
+    assert hasattr(app.state, "sentiment_model")
+    assert hasattr(app.state, "lookup_tables")
     assert app.state.sentiment_model == "mock_roberta_sentiment_model"
 
     # Verify lookup tables structure matches template