カスタムユーザー保存領域

カスタムユーザー保存領域

お知らせ：当社は、お客様により充実したサポート情報を迅速に提供するため、本ページのコンテンツは機械翻訳を用いて日本語に翻訳しています。正確かつ最新のサポート情報をご覧いただくには、本内容の英語版を参照してください。

概要  

デフォルトでは、SDKは訪問者へのバリエーションの割り当てをメモリに保持します。本番環境のアプリケーションでは、add_custom_storage()を使用してカスタムストレージの実装を指定し、アプリケーションの再起動後も訪問者への割り当てを保持できます。永続ストレージを使用すると、アプリケーションの再起動後も訪問者に同じバリエーションが提供されます。

メソッド  

add_custom_storage(storage: UserStorageService)

パラメーター  

パラメーター
種類
必須
説明
storage
UserStorageService
はい
カスタムユーザーストレージの実装。

戻り値  

種類
説明
PageSenseSDKOptions
メソッドチェーンを使用できるように、同じSDKオプションのインスタンスを返します。

使用例  

from pagesense import PageSenseSDKOptions
 
# Create your custom UserStorageService implementation
custom_storage = MyUserStorageService()
 
sdk_options = (
PageSenseSDKOptions()
.add_custom_storage(custom_storage)
)
 
カスタムユーザーストレージクラスでは、UserStorageServiceのインターフェースを実装し、次のメソッドを実装する必要があります。
 
  • lookup() – 訪問者について以前保存されたバリエーションの割り当てを取得します。
  • save() – 今後のリクエストに備えて、訪問者へのバリエーションの割り当てを永続化します。
 
SDKは、実験の評価中にこれらのメソッドを自動的に呼び出し、訪問者への割り当てを維持します。
   
実装例
 
次の例は、UserStorageServiceのインターフェースの実装例です。データベース、ファイルシステム、分散キャッシュなど、アプリケーションで使用する永続ストレージの仕組みと連携できるように実装を変更できます。
  
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