How to Mock a Feature Store Online Lookup in Python
This code simulates an online feature store with single and batch retrieval methods, using a dict-backed cache and timestamps.
Python code
51 linesimport random
import time
class OnlineFeatureStore:
def __init__(self):
self.features = {}
def put(self, entity_id: str, feature_name: str, value):
key = (entity_id, feature_name)
self.features[key] = (value, time.time())
def get(self, entity_id: str, feature_name: str):
key = (entity_id, feature_name)
if key not in self.features:
raise KeyError(f"Feature {feature_name!r} not found for entity {entity_id!r}")
value, timestamp = self.features[key]
return {"value": value, "timestamp": timestamp}
def batch_get(self, entity_ids, feature_names):
results = {}
for entity_id in entity_ids:
results[entity_id] = {}
for feature_name in feature_names:
try:
results[entity_id][feature_name] = self.get(entity_id, feature_name)
except KeyError:
results[entity_id][feature_name] = None
return results
if __name__ == "__main__":
store = OnlineFeatureStore()
store.put("user_123", "age", 34)
store.put("user_123", "city", "Berlin")
store.put("user_456", "age", 28)
store.put("user_456", "city", "Paris")
print("Single lookup:")
print(store.get("user_123", "city"))
print("\nBatch lookup:")
batch_result = store.batch_get(
entity_ids=["user_123", "user_456"],
feature_names=["age", "city", "premium"],
)
for entity, features in batch_result.items():
print(f"{entity}:")
for feature, data in features.items():
print(f" {feature}: {data}")
Output
Single lookup:
{'value': 'Berlin', 'timestamp': 1712345678.1234567}
Batch lookup:
user_123:
age: {'value': 34, 'timestamp': 1712345678.1234567}
city: {'value': 'Berlin', 'timestamp': 1712345678.1234567}
premium: None
user_456:
age: {'value': 28, 'timestamp': 1712345678.1234567}
city: {'value': 'Paris', 'timestamp': 1712345678.1234567}
premium: None
How it works
The OnlineFeatureStore class stores feature values in a plain dictionary with a tuple (entity_id, feature_name) as the key. Each entry keeps the value and a Unix timestamp from time.time(), simulating when the feature was written. The get method returns a dict with both the value and timestamp, or raises a KeyError for missing features, matching real feature store behavior. batch_get loops over all combinations of entity IDs and feature names, catching missing features and filling them with None so callers always get a consistent structure. This pattern mirrors production feature stores like Feast or Tecton, where online lookups must be fast and return structured results for ML inference.
Common mistakes
- Returning just the value instead of a dict with timestamp, breaking downstream code expecting metadata.
- Not raising a clear KeyError for missing features, making debugging harder in production.
- Using a mutable default for batch_get (e.g., `def batch_get(self, entity_ids, feature_names=[...])`) which persists state across calls.
- Overwriting timestamps too frequently, causing misleading freshness signals in analytics.
Variations
- Use a `defaultdict` with a sentinel value to avoid explicit None checks.
- Implement a TTL-based eviction to mimic real feature store expiry of stale values.
Real-world use cases
- Feeding live user features (e.g., age, location) to an ML model for real-time recommendations or personalization.
- Serving training and inference features in a low-latency path for fraud detection or risk scoring.
- Providing a lightweight local mock of a feature store when developing or unit-testing ML pipelines without a cloud dependency.
Sponsored
More from ML engineering pipelines
- Bayesian Optimization in Python: A Simplified Mock Implementation medium
- Build a Data Helper Class in Python for ML Pipelines easy
- Build a Mock Random Forest Classifier in Python easy
- Champion Challenger Deployment Mock in Python easy
- Compare Model A vs Model B Metrics in Python easy
- Create a Minimal Great Expectations Suite Mock in Python easy
Keep learning
Related tutorials and quizzes for this topic.