Custom User Storage

Overview

By default, the SDK maintains visitor assignments in memory. For production applications, you can provide a custom storage implementation using add_custom_storage() to persist visitor assignments across application restarts. Persistent storage helps ensure that visitors continue to receive the same variation even after the application is restarted.

Method

add_custom_storage(storage: UserStorageService)

Parameters

Parameter	Type	Required	Description
storage	UserStorageService	Yes	Custom user storage implementation.

Return Value

Type	Description
PageSenseSDKOptions	Returns the same SDK options instance to support method chaining.

Usage Example

from pagesense import PageSenseSDKOptions
 
# Create your custom UserStorageService implementation
custom_storage = MyUserStorageService()
 
sdk_options = (
    PageSenseSDKOptions()
        .add_custom_storage(custom_storage)
)

Your custom user storage class must implement the UserStorageService interface and provide implementations for the following methods:

lookup() – Retrieves the previously stored variation assignment for a visitor.
save() – Persists the visitor's variation assignment for future requests.

The SDK automatically invokes these methods during experiment evaluation to maintain sticky visitor assignment.

Sample Implementation

The following example demonstrates a sample implementation of the UserStorageService interface. You can modify the implementation to integrate with your application's preferred persistent storage mechanism, such as a database, file system, or distributed cache.

import os
import json
import logging
import asyncio
 
from pagesense.storage.userStorageService import UserStorageService
 
"""
 Implements the UserStorageService interface that writes the user profile details to a local JSON file
 to store all the users' experiment to variation mappings. This is a simple, file-based approach that
 can be replaced with other persistent storage mechanisms like databases, Redis, or cloud storage.
"""
class FileUserStorageService(UserStorageService):
 
    USER_DATA_FILE_NAME = 'user_storage.json'
    USER_LOG_FILE_NAME = 'user_storage.log'
 
    def __init__(self, storage_directory):
        """
        Initialize a new FileUserStorageService instance.
 
        Args:
            storageDirectory: Path where the user storage & log files will be created.
 
        Returns:
            None
        """
        # Call the Super Constructor
        super().__init__()
 
        self.logger = None
        self.storage_directory = None
        self.user_data_file_path = None
        self.user_log_file_path = None
 
        # Create and verify the storage directory.
        self.create_storage_directory(storage_directory)
 
        # Create the user storage file if missing.
        self.create_user_data_file()
 
        # Initialize the logger after storage setup is complete.
        self.initialize_logger()
 
        # Log the successful setup.
        self.logger.info('FileUserStorageService initialized successfully.', extra={
            "storageDirectory": self.storage_directory,
            "userDataFilePath": self.user_data_file_path,
            "userLogFilePath": self.user_log_file_path
        })
 
    def create_storage_directory(self, storage_directory):
        """
        Create and validate the storage directory.
 
        Args:
            storageDirectory: Path where the user storage & log files will be stored.
 
        Returns:
            None
        """
        # If directory does not exist, attempt to create it recursively.
        if not os.path.exists(storage_directory):
            try:
                os.makedirs(storage_directory, exist_ok=True)
            except:
                raise Exception(f"Failed to create the storage directory: {storage_directory}")
 
        # Ensure the directory is writable to avoid runtime failures.
        if not os.access(storage_directory, os.W_OK):
            raise Exception(f"Storage directory is not writable: {storage_directory}")
 
        # Define the directory and the file paths for internal use.
        self.storage_directory = storage_directory
        self.user_data_file_path = os.path.join(storage_directory, self.USER_DATA_FILE_NAME)
        self.user_log_file_path = os.path.join(storage_directory, self.USER_LOG_FILE_NAME)
 
    def create_user_data_file(self):
        """
        Create the user storage JSON file if it does not exist.
 
        Args:
            None
 
        Returns:
            None
        """
        # Check if the user storage file exists.
        if not os.path.exists(self.user_data_file_path):
            try:
                with open(self.user_data_file_path, 'w') as f:
                    json.dump({}, f, indent=2)
            except:
                raise Exception(f"Failed to create the user data file at: {self.user_data_file_path}")
 
    def initialize_logger(self):
        """
        Initialize the logger for file-based logging.
 
        Args:
            None
 
        Returns:
            None
        """
        # Ensure the log file exists
        if not os.path.exists(self.user_log_file_path):
            with open(self.user_log_file_path, 'w'):
                pass
 
        # Configure the logger
        self.logger = logging.getLogger("FileUserStorageService")
        self.logger.setLevel(logging.DEBUG)
 
        file_handler = logging.FileHandler(self.user_log_file_path)
        file_handler.setLevel(logging.DEBUG)
 
        formatter = logging.Formatter(
            '%(asctime)s %(levelname)s %(message)s'
        )
 
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)
 
        # Log successful logger creation
        self.logger.info('Logger initialized successfully.', extra={
            "logFile": self.user_log_file_path
        })
 
    def look_up(self, user_id):
        """
        Look up the experiment–variation mapping for a user.
 
        Args:
            userId: Unique user identifier.
 
        Returns:
            JSON string of user mappings or None
        """
        self.logger.info('Looking up user data.', extra={"userId": user_id})
 
        # Verify that the user storage file exists.
        if not os.path.exists(self.user_data_file_path):
            self.logger.warning('User data file not found.', extra={"userId": user_id})
            return None
 
        # Attempt to read the file contents.
        with open(self.user_data_file_path, 'r') as f:
            file_contents = f.read()
 
        # Decode JSON safely
        decoded_data = self.safe_json_decode(file_contents)
 
        # Verify JSON structure
        if not decoded_data or not isinstance(decoded_data, dict):
            self.logger.warning('Invalid JSON format.', extra={"path": self.user_data_file_path})
            return None
 
        if user_id not in decoded_data:
            self.logger.info('No mappings found.', extra={"userId": user_id})
            return None
 
        self.logger.info('Mappings retrieved.', extra={"userId": user_id})
 
        return json.dumps(decoded_data[user_id])
 
    def save(self, user_profile):
        """
        Save or update user experiment–variation mapping.
 
        Args:
            userProfile: UserProfile instance.
 
        Returns:
            None
        """
        user_id = user_profile.get_user_id()
 
        self.logger.info('Saving user data.', extra={"userId": user_id})
 
        try:
            # Decode incoming user profile JSON
            user_profile_data = self.safe_json_decode(user_profile.get_user_profile_json_string())
 
            # Validate data
            if not user_profile_data or not isinstance(user_profile_data, dict):
                self.logger.error('Invalid user profile JSON.', extra={"userId": user_id})
                return
 
            # Load existing data
            try:
                with open(self.user_data_file_path, 'r') as f:
                    file_contents = f.read()
            except:
                file_contents = '{}'
 
            existing_data = self.safe_json_decode(file_contents) or {}
 
            # Update user data
            existing_data[user_id] = user_profile_data
 
            # Write updated data
            with open(self.user_data_file_path, 'w') as f:
                json.dump(existing_data, f, indent=2)
 
            self.logger.info('User data saved.', extra={
                "userId": user_id,
                "file": self.user_data_file_path
            })
 
        except Exception as err:
            self.logger.error('Failed to save user data.', extra={
                "userId": user_id,
                "path": self.user_data_file_path,
                "message": str(err)
            })
 
    def safe_json_decode(self, json_string):
        """
        Safely decode a JSON string.
 
        Args:
            jsonString: JSON string to decode.
 
        Returns:
            Decoded object or None
        """
        # Validate input
        if not json_string or not json_string.strip():
            return None
 
        try:
            return json.loads(json_string)
        except Exception as err:
            if self.logger:
                self.logger.warning('JSON decoding failed.', extra={
                    "errorMessage": str(err)
                })
return None