feat: add support for vwo_feTrackUsage event

Author: Kaustubh22327

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.55.0] - 2026-06-16
+
+### Added
+
+- Added support for FE Usage Tracking. The SDK now accurately tracks usage for features by firing a new usage tracking event (`vwo_feTrackUsage`) for storage hits, feature-not-found cases, and cached holdouts etc.
+
 ## [1.50.0] - 2026-05-28
 
 This release introduces **Wingify** as the primary SDK branding and package namespace, while keeping existing **VWO** integrations fully supported.

setup.py

@@ -143,7 +143,7 @@ def run(self):
 
 setup(
     name=_brand["name"],
-    version="1.50.0",
+    version="1.55.0",
     description=_brand["description"],
     long_description=long_description,
     long_description_content_type="text/markdown",

wingify/api/get_flag_api.py

@@ -36,6 +36,7 @@
 from ..utils.impression_util import (
     send_impression_for_variation_shown_batch,
     send_impression_for_variation_shown,
+    create_and_send_impression_for_usage_tracking,
 )
 from ..models.user.get_flag import GetFlag
 from ..services.storage_service import StorageService
@@ -76,6 +77,9 @@ def get(
         :param hook_manager: The hook manager to execute hooks.
         """
 
+        # Flag for usage tracking - false if no variation shown call is sent
+        is_variation_shown_fired = False
+
         feature = get_feature_from_key(settings, feature_key)
         is_enabled = False
         variables = []
@@ -173,6 +177,15 @@ def get(
                                 )
                         else:
                             send_impression_for_variation_shown_batch(holdout_payloads, account_id=settings.get_account_id(), sdk_key=settings.get_sdk_key())
+                        
+                        # if impression is fired for any of the holdouts, set is_variation_shown_fired to True
+                        if holdout_payloads and len(holdout_payloads) > 0:
+                            is_variation_shown_fired = True
+
+                        # Case: User is already part of a holdout (stored decision).
+                        if settings.get_is_tracking_usage_enabled() and not is_variation_shown_fired:
+                            create_and_send_impression_for_usage_tracking(settings, feature_key, context)
+
                         # Return a disabled flag as user is in persistent holdout
                         return GetFlag(is_enabled=False, variables=[], session_id=context.get_session_id(), uuid=context.get_vwo_uuid())
         if stored_data and stored_data.get("experimentVariationId"):
@@ -194,7 +207,19 @@ def get(
                     )
                     decision["isUserPartOfCampaign"] = True
                     # send network calls for holdouts that are newly added in settings and are not present in storage
-                    send_network_calls_for_not_in_holdouts(settings, feature, context, decision, stored_data, storage_service)
+                    # Calculate the initial size of not_in_holdout_id
+                    initial_size = len(stored_data.get("notInHoldoutId") or []) if stored_data else 0
+                    # send network calls for not_in_holdout_id
+                    updated_not_in_holdouts = send_network_calls_for_not_in_holdouts(settings, feature, context, decision, stored_data, storage_service)
+                    # if variationShown event is fired, set is_variation_shown_fired to True - Optimization for newly added holdouts
+                    if updated_not_in_holdouts and len(updated_not_in_holdouts) > initial_size:
+                        is_variation_shown_fired = True
+
+                    # Case: Feature found in storage
+                    # Send usage tracking for cached experiment decision if usage tracking is enabled
+                    if settings.get_is_tracking_usage_enabled() and not is_variation_shown_fired:
+                        create_and_send_impression_for_usage_tracking(settings, feature_key, context)
+
                     return GetFlag(is_enabled=True, variables=variation.get_variables(), session_id=context.get_session_id(), uuid=context.get_vwo_uuid())
         elif (
             stored_data
@@ -217,7 +242,10 @@ def get(
                 decision["isUserPartOfCampaign"] = True
                 # network calls for holdouts that are newly added in settings and are not present in storage
                 # and return the updated not in holdout ids
+                initial_size = len(stored_data.get("notInHoldoutId") or []) if stored_data else 0
                 updated_not_in_holdout_ids = send_network_calls_for_not_in_holdouts(settings, feature, context, decision, stored_data, storage_service)
+                if updated_not_in_holdout_ids and len(updated_not_in_holdout_ids) > initial_size:
+                    is_variation_shown_fired = True
                 # push the updated not in holdout ids to the notInHoldoutIds array
                 notInHoldoutIds.extend(updated_not_in_holdout_ids)
 
@@ -236,6 +264,12 @@ def get(
         if feature is None:
             LogManager.get_instance().error_log("FEATURE_NOT_FOUND", data={"featureKey": feature_key}, debug_data = debug_event_props)
             is_enabled = False
+
+            # Case: Feature not found
+            # If usage tracking is enabled, send usage tracking impression
+            if settings.get_is_tracking_usage_enabled() and not is_variation_shown_fired:
+                create_and_send_impression_for_usage_tracking(settings, feature_key, context)
+
             return GetFlag(is_enabled=is_enabled, variables=variables, session_id=context.get_session_id(), uuid=context.get_vwo_uuid())
 
         if context.get_session_id() is None:
@@ -315,6 +349,9 @@ def get(
                 else:
                     batchPayload.extend(holdout_payloads)
 
+                if holdout_payloads and len(holdout_payloads) > 0:
+                    is_variation_shown_fired = True
+
         roll_out_rules = get_specific_rules_based_on_type(
             feature, CampaignTypeEnum.ROLLOUT.value
         )
@@ -390,9 +427,13 @@ def get(
                         and payload is not None
                         and len(payload) > 0
                     ):
+                        # set is_variation_shown_fired to true as the rollout impression is sent
+                        is_variation_shown_fired = True
                         send_impression_for_variation_shown(payload, context)
                     else:
                         if payload is not None and len(payload) > 0:
+                            # set is_variation_shown_fired to true as the rollout impression is sent
+                            is_variation_shown_fired = True
                             batchPayload.append(payload)
 
         if not roll_out_rules:
@@ -435,9 +476,15 @@ def get(
                             and payload is not None
                             and len(payload) > 0
                         ):
+                            # set is_variation_shown_fired to true as the experiment impression is sent (for whitelisted user)
+                            # this prevents sending usage tracking impression for the experiment campaign when checked at the bottom
+                            is_variation_shown_fired = True
                             send_impression_for_variation_shown(payload, context)
                         else:
                             if payload is not None and len(payload) > 0:
+                                # set is_variation_shown_fired to true as the experiment impression is sent (for whitelisted user)
+                                # this prevents sending usage tracking impression for the experiment campaign when checked at the bottom
+                                is_variation_shown_fired = True
                                 batchPayload.append(payload)
 
                         is_enabled = True
@@ -481,9 +528,13 @@ def get(
                         and payload is not None
                         and len(payload) > 0
                     ):
+                        # set is_variation_shown_fired to true as the experiment impression is sent.
+                        is_variation_shown_fired = True
                         send_impression_for_variation_shown(payload, context)
                     else:
                         if payload is not None and len(payload) > 0:
+                            # set is_variation_shown_fired to true as the experiment impression is sent.
+                            is_variation_shown_fired = True
                             batchPayload.append(payload)
 
         if is_enabled:
@@ -545,16 +596,25 @@ def get(
                 and payload is not None
                 and len(payload) > 0
             ):
+                # set is_variation_shown_fired to true as the impact analysis impression is sent.
+                is_variation_shown_fired = True
                 send_impression_for_variation_shown(payload, context)
             else:
                 if payload is not None and len(payload) > 0:
+                    # set is_variation_shown_fired to true as the impact analysis impression is sent.
+                    is_variation_shown_fired = True
                     batchPayload.append(payload)
 
         if not SettingsManager.get_instance().is_gateway_service_provided:
             send_impression_for_variation_shown_batch(
                 batchPayload, settings.get_account_id(), settings.get_sdk_key()
             )
 
+        # Send usage tracking call when no primary variationShown event was dispatched.
+        # If a primary event was fired, the server already has the usage tracking signal.
+        if settings.get_is_tracking_usage_enabled() and not is_variation_shown_fired:
+            create_and_send_impression_for_usage_tracking(settings, feature_key, context)
+
         return GetFlag(is_enabled=is_enabled, variables=variables, session_id=context.get_session_id(), uuid=context.get_vwo_uuid())
 
     def _update_integrations_decision_object(

wingify/constants/Constants.py

@@ -17,7 +17,7 @@ class Constants:
     # TODO: read from setup.py
     sdk_meta = {
         "name": "vwo-fme-python-sdk",
-        "version": "1.50.0",
+        "version": "1.55.0",
     }
 
     SDK_VERSION = sdk_meta["version"]

wingify/enums/event_enum.py

@@ -23,3 +23,4 @@ class EventEnum(Enum):
     SDK_INIT_EVENT = "vwo_fmeSdkInit"
     USAGE_STATS = "vwo_sdkUsageStats"
     DEBUGGER_EVENT = "vwo_sdkDebug"
+    TRACK_USAGE = "vwo_feTrackUsage"

wingify/models/settings/settings_model.py

@@ -36,6 +36,7 @@ def __init__(self, data: Dict):
         self._poll_interval = data.get("pollInterval", Constants.POLLING_INTERVAL)
         self._is_web_connectivity_enabled = data.get("isWebConnectivityEnabled", True)
         self._holdouts = [HoldoutModel(h) for h in data.get("holdouts", [])]
+        self._is_tracking_usage_enabled = data.get("isMAU", False)
 
     # Getter methods for accessing private attributes
     def get_features(self) -> List[FeatureModel]:
@@ -100,4 +101,10 @@ def set_holdouts(self, value: List[HoldoutModel]):
         self._holdouts = value
 
     def get_holdouts(self) -> List[HoldoutModel]:
-        return self._holdouts if isinstance(self._holdouts, list) else []
+        return self._holdouts if isinstance(self._holdouts, list) else []
+
+    def get_is_tracking_usage_enabled(self) -> bool:
+        return self._is_tracking_usage_enabled
+
+    def set_is_tracking_usage_enabled(self, value: bool):
+        self._is_tracking_usage_enabled = value

wingify/resources/debug-messages.json

@@ -16,5 +16,6 @@
     "HOLDOUT_SEGMENTATION_FAIL": "User ID: {userId} does not qualify segmentation for Holdout: {holdoutGroupName}",
     "HOLDOUT_SHOULD_NOT_EXCLUDE_USER": "User ID: {userId} is not part of Holdout: {holdoutGroupName} for feature: {featureKey}",
     "PART_OF_HOLDOUT_IN_MEG": "MEG: User ID:{userId} is part of holdout: {holdoutId} for feature: {featureKey}",
-    "HOLDOUT_SKIP_EVALUATION": "Skip holdout reevaluation for holdout ID: {holdoutId} because {reason}"
+    "HOLDOUT_SKIP_EVALUATION": "Skip holdout reevaluation for holdout ID: {holdoutId} because {reason}",
+    "IMPRESSION_FOR_TRACK_USAGE": "Impression built for vwo_feTrackUsage({brand} standard event for tracking usage) event having Account ID:{accountId}, User ID:{userId}, and feature key:{featureKey}"
   }

wingify/utils/impression_util.py

@@ -80,4 +80,72 @@ def send_batch_request():
             network_instance.execute_in_background(send_batch_request)
         else:
             # execute the request immediately
-            send_batch_request()
+            send_batch_request()
+
+
+def create_and_send_impression_for_usage_tracking(
+    settings, feature_key: str, context: ContextModel
+):
+    """
+    Creates and sends an impression for usage tracking.
+    
+    This function is used to send a tracking event to the server when a feature flag
+    is evaluated but no primary `variation_shown` event is dispatched (e.g., when serving
+    cached decisions from storage, or when the flag is not found).
+    
+    :param settings: The settings file containing the account settings.
+    :param feature_key: The feature key for which usage is being tracked.
+    :param context: The ContextModel object containing user and session context.
+    """
+    from ..wingify_client import WingifyClient as VWOClient
+    from ..utils.network_util import _get_event_base_payload
+    from ..packages.logger.core.log_manager import LogManager
+    from ..utils.log_message_util import debug_messages
+    from ..utils.function_util import get_current_unix_timestamp_in_millis
+    from ..utils.brand_util import get_sdk_name
+    from ..services.settings_manager import SettingsManager
+
+    user_id = context.get_id()
+
+    properties = _get_event_base_payload(
+        settings=settings,
+        user_id=user_id,
+        event_name=EventEnum.TRACK_USAGE.value,
+        visitor_user_agent=context.get_user_agent(),
+        ip_address=context.get_ip_address(),
+    )
+
+    if context.get_session_id() is not None and context.get_session_id() != 0:
+        properties["d"]["sessionId"] = context.get_session_id()
+
+    # if uuid is provided in the context, use it, otherwise generate a new one
+    if context.get_vwo_uuid() is not None and context.get_vwo_uuid() != "":
+        properties["d"]["msgId"] = f"{context.get_vwo_uuid()}-{get_current_unix_timestamp_in_millis()}"
+        properties["d"]["visId"] = context.get_vwo_uuid()
+
+    LogManager.get_instance().debug(
+        debug_messages.get("IMPRESSION_FOR_TRACK_USAGE").format(
+            brand=get_sdk_name(SettingsManager.get_instance().is_via_vwo),
+            accountId=settings.get_account_id(),
+            userId=user_id,
+            featureKey=feature_key
+        )
+    )
+
+    # Note: id, variation, isFirst are not set!
+
+    vwo_instance = VWOClient.get_instance()
+    
+    # Get base properties for the event
+    base_properties = get_events_base_properties(
+        EventEnum.TRACK_USAGE.value,
+        visitor_user_agent=context.get_user_agent(),
+        ip_address=context.get_ip_address(),
+    )
+
+    if vwo_instance.batch_event_queue is not None:
+        # Enqueue the event to the batch queue
+        vwo_instance.batch_event_queue.enqueue(properties)
+    else:
+        # Send the event immediately if batch events are not enabled
+        send_post_api_request(base_properties, properties, user_id)