Experiment Python SDK
Official documentation for Amplitude Experiment's server-side Python SDK implementation.
This documentation has separate sections for remote and local evaluation.
Remote evaluation
Implements fetching variants for a user using remote evaluation.
Install
Python version compatibility
The Python Server SDK works with Python 3.6+.
Install the Python Server SDK with pip.
pip install amplitude-experiment
Quick start
from amplitude_experiment import Experiment, RemoteEvaluationConfig, RemoteEvaluationClient, User
# (1) Initialize the experiment client
experiment = Experiment.initialize_remote('<DEPLOYMENT_KEY>')
# (2) Fetch variants for a user
user = User(
device_id="abcdefg",
user_id="user@company.com",
user_properties={
'premium': True
}
)
variants = experiment.fetch_v2(user)
# (3) Access a flag's variant
variant = variants['YOUR-FLAG-KEY']
if variant:
if variant.value == 'on':
# Flag is on
else:
# Flag is off
Initialize
Initialize the SDK client in your server on startup. The deployment key argument you pass to the api_key parameter must live within the same project that you are sending analytics events to.
Experiment.initialize_remote(api_key, config = None) : RemoteEvaluationClient
| Parameter | Requirement | Description |
|---|---|---|
api_key | required | The deployment key that authorizes fetch requests and determines which flags to evaluate for the user. |
config | optional | The client configuration used to customize SDK client behavior. |
Timeout and retry configuration
Configure the timeout and retry options to best fit your performance requirements.
experiment = Experiment.initialize_remote('<DEPLOYMENT_KEY>', Config())
Configuration
You can configure the SDK client on initialization.
EU data center
If you're using Amplitude's EU data center, configure the server_zone option on initialization.
| Name | Description | Default Value |
|---|---|---|
debug | When True, sets the logger level to DEBUG. | False |
logger | Custom logging.Logger instance for SDK logging. | Default Logger with WARNING level |
server_zone | The Amplitude data center to use. Either ServerZone.US or ServerZone.EU | ServerZone.US |
server_url | The host to fetch variants from. | https://api.lab.amplitude.com |
fetch_timeout_millis | The timeout for fetching variants in milliseconds. This timeout only applies to the initial request, not subsequent retries | 10000 |
fetch_retries | The number of retries to attempt if a request to fetch variants fails. | 0 |
fetch_retry_backoff_min_millis | The minimum (initial) backoff after a request to fetch variants fails. The SDK scales this delay by the fetchRetryBackoffScalar | 500 |
fetch_retry_backoff_max_millis | The maximum backoff between retries. If the scaled backoff becomes greater than the max, the SDK uses the max for all subsequent requests | 10000 |
fetch_retry_backoff_scalar | Scales the minimum backoff exponentially. | 1.5 |
fetch_retry_timeout_millis | The request timeout for retrying variant fetches. | 10000 |
Fetch
Fetches variants for a user and returns the results. The function remote evaluates the user for flags associated with the deployment used to initialize the SDK client.
fetch_v2(user: User, fetch_options: FetchOptions = None) : Variants
FetchOptions
| Name | Description | Default Value |
|---|---|---|
tracks_exposure | To track or not track an exposure event for this fetch request. If None, uses the server's default behavior (doesn't track exposure). | None |
tracks_assignment | To track or not track an assignment event for this fetch request. If None, uses the server's default behavior (does track assignment). | None |
user = User(
device_id="abcdefg",
user_id="user@company.com",
user_properties={
'premium': True
}
)
variants = experiment.fetch_v2(user)
After fetching variants for a user, you may to access the variant for a specific flag.
variant = variants['YOUR-FLAG-KEY']
if variant:
if variant.value == 'on':
# Flag is on
else:
# Flag is off
Fetch async
The fetch method is synchronous. To fetch asynchronously, you can use fetch_async method
fetch_async_v2(user: User, callback)
| Parameter | Requirement | Description |
|---|---|---|
user | required | The user for whom to fetch variants. |
callback | optional | The callback to handle the variants. |
def fetch_callback(user, variants):
variant = variants['YOUR-FLAG-KEY']
if variant:
if variant.value == 'on':
# Flag is on
else:
# Flag is off
experiment.fetch_async_v2(user, fetch_callback)
Local evaluation
Implements evaluating variants for a user using local evaluation. If you plan to use local evaluation, you should understand the tradeoffs.
Install
Install the Python Server SDK's local evaluation.
Operating system and architecture support
The local evaluation package supports the following OS' and architectures (OS/ARCH):
Supported
- darwin/amd64
- darwin/arm64
- linux/amd64
- linux/arm64
The local evaluation package doesn't support Alpine Linux.
If you need another OS/Arch supported, submit an issue on github or email experiment@amplitude.com.
Install the Python Server SDK with pip.
pip install amplitude-experiment
Quick start
# (1) Initialize the local evaluation client with a server deployment key.
experiment = Experiment.initialize_local("DEPLOYMENT_KEY", LocalEvaluationConfig(
# (Recommended) Enable local evaluation cohort targeting.
cohort_sync_config=CohortSyncConfig(api_key="API_KEY", secret_key="SECRET_KEY")
))
# (2) Start the local evaluation client.
experiment.start()
# (3) Evaluate a user.
user = User(
device_id="abcdefg",
user_id="user@company.com",
user_properties={
'premium': True
}
)
variants = experiment.evaluate_v2(user)
Initialize
Initializes a local evaluation client.
Server deployment key
You must initialize the local evaluation client with a server deployment key in to get access to local evaluation flag configs.
Experiment.initialize_local(api_key, config = None) : LocalEvaluationClient
| Parameter | Requirement | Description |
|---|---|---|
apiKey | required | The server deployment key that authorizes fetch requests and determines which flags to evaluate for the user. |
config | optional | The client configuration used to customize SDK client behavior. |
Flag polling interval
Use the flag_config_polling_interval_millis configuration to determine the time flag configs take to update after modification (default 30s).
Configuration
You can configure the SDK client on initialization.
EU data center
If you're using Amplitude's EU data center, configure the server_zone option on initialization.
LocalEvaluationConfig
| Name | Description | Default Value |
|---|---|---|
debug | When True, sets the logger level to DEBUG. | False |
logger | Custom logging.Logger instance for SDK logging. | Default Logger with WARNING level |
server_zone | The Amplitude data center to use. Either ServerZone.US or ServerZone.EU | ServerZone.US |
server_url | The host to fetch flag configurations from. | https://api.lab.amplitude.com |
flag_config_polling_interval_millis | The interval to poll for updated flag configs after calling start() | 30000 |
flag_config_poller_request_timeout_millis | The timeout for the request made by the flag config poller | 10000 |
assignment_config | Deprecated. Use exposure_config instead. Configuration for automatically tracking assignment events after an evaluation. | None |
exposure_config | Configuration for tracking exposure events after an evaluation. | None |
cohort_sync_config | Configuration to enable cohort downloading for local evaluation cohort targeting. | None |
AssignmentConfig
| Name | Description | Default Value |
|---|---|---|
api_key | The analytics API key and NOT the experiment deployment key | required |
cache_capacity | The maximum number of assignments stored in the assignment cache | 65536 |
send_evaluated_props | Set to True to send properties of the evaluated user in assignment events | False |
| Analytics SDK Options | Options to configure the underlying Amplitude Analytics SDK used to track assignment events |
ExposureConfig
| Name | Description | Default Value |
|---|---|---|
api_key | The analytics API key. This is not the experiment deployment key | required |
cache_capacity | The maximum number of exposures stored in the exposure cache | 65536 |
| Analytics SDK Options | Options to configure the underlying Amplitude Analytics SDK used to track exposure events |
CohortSyncConfig
| Name | Description | Default Value |
|---|---|---|
api_key | The analytics API key and NOT the experiment deployment key | required |
secret_key | The analytics secret key | required |
max_cohort_size | The maximum size of cohort that the SDK downloads. The SDK doesn't download cohorts larger than this size. | 2147483647 |
cohort_polling_interval_millis | The interval, in milliseconds, to poll Amplitude for cohort updates (60000 minimum). | 60000 |
cohort_server_url | The cohort server endpoint from which to fetch cohort data. For hitting the EU data center, set server_zone to ServerZone.EU. Setting this value overrides server_zone defaults. | https://cohort-v2.lab.amplitude.com |
Start
Start the local evaluation client, pre-fetching local evaluation mode flag configs for evaluation and starting the flag config poller at the configured interval.
start()
You should await the result of start() to ensure that flag configs are ready to be used before calling evaluate_v2()
experiment.start()
Evaluate
Executes the evaluation logic using the flags pre-fetched on start(). You must give evaluate a user object argument. You can optionally pass an array of flag keys if you require only a specific subset of flag variants.
Automatic assignment tracking
Set assignment_config to automatically track an assignment event to Amplitude when evaluate_v2() is called.
Exposure tracking
Set exposure_config to enable exposure tracking. Then, set tracks_exposure to True in EvaluateOptions when calling evaluate_v2().
evaluate_v2(self, user: User, flag_keys: List[str], options: EvaluateOptions) : Dict[str, Variant]
# The user to evaluate
user = User(user_id='test_user')
# Evaluate all flag variants
all_variants = experiment.evaluate_v2(user)
# Evaluate a specific subset of flag variants
specific_variants = experiment.evaluate_v2(user, ["<FLAG_KEY_1>", "<FLAG_KEY_2>"])
# Access a variant
variant = all_variants["<FLAG_KEY>"]
if variant.value == 'on':
# Flag is on
else:
# Flag is off
EvaluateOptions
| Name | Description | Default Value |
|---|---|---|
tracks_exposure | If True, the SDK tracks an exposure event for the evaluated variants. | False |
Local evaluation cohort targeting
Since version 1.4.0, the local evaluation SDK client supports downloading cohorts for local evaluation targeting. You must configure the cohort_sync_config option with the analytics api_key and secret_key on initialization to enable this support.
experiment = Experiment.initialize_local("DEPLOYMENT_KEY", LocalEvaluationConfig(
# (Recommended) Enable local evaluation cohort targeting.
cohort_sync_config=CohortSyncConfig(api_key="API_KEY", secret_key="SECRET_KEY")
))
Custom logging
Pass a custom logging.Logger instance to control logging behavior.
Custom logger
Pass a custom logging.Logger instance to RemoteEvaluationConfig or LocalEvaluationConfig:
import logging
from amplitude_experiment import Experiment, RemoteEvaluationConfig, LocalEvaluationConfig
# Create a custom logger
custom_logger = logging.getLogger('MyAppLogger')
custom_logger.setLevel(logging.WARN)
handler = logging.FileHandler('experiment.log')
handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
custom_logger.addHandler(handler)
# Remote evaluation with custom logger
remote_config = RemoteEvaluationConfig(
logger=custom_logger
)
experiment = Experiment.initialize_remote('DEPLOYMENT_KEY', remote_config)
# Local evaluation with custom logger
local_config = LocalEvaluationConfig(
logger=custom_logger
)
experiment = Experiment.initialize_local('DEPLOYMENT_KEY', local_config)
Debug flag with default logger
Without a custom logger, the debug flag controls the default logger's level:
# Without custom logger, debug=False uses WARNING level
config = RemoteEvaluationConfig(
debug=False
)
# Default logger level is WARNING
# Without custom logger, debug=True uses DEBUG level
config = RemoteEvaluationConfig(
debug=True
)
# Default logger level is DEBUG
With a custom logger, the SDK ignores the debug flag and your logger maintains its configured level:
import logging
from amplitude_experiment import Experiment, RemoteEvaluationConfig
custom_logger = logging.getLogger('MyAppLogger')
custom_logger.setLevel(logging.WARN)
# Custom logger maintains its WARN level regardless of debug flag
config = RemoteEvaluationConfig(
logger=custom_logger,
debug=True
)
# Logger level stays WARN (debug flag is ignored)
Access Amplitude cookies
If you're using the Amplitude Analytics SDK on the client-side, the Python server SDK provides an AmplitudeCookie class with convenience functions for parsing and interacting with the Amplitude identity cookie. This is useful for ensuring that the Device ID on the server matches the Device ID set on the client, especially if the client hasn't yet generated a Device ID.
import uuid
from amplitude_experiment import AmplitudeCookie
# Get the cookie name for the Amplitude API key
# For Browser SDK 2.0 cookies, use new_format=True:
# amp_cookie_name = AmplitudeCookie.cookie_name('amplitude-api-key', new_format=True)
amp_cookie_name = AmplitudeCookie.cookie_name('amplitude-api-key')
device_id = None
# Try to get device ID from existing cookie
if request.cookies.get(amp_cookie_name):
device_id = AmplitudeCookie.parse(request.cookies.get(amp_cookie_name)).device_id
# For Browser SDK 2.0: AmplitudeCookie.parse(request.cookies.get(amp_cookie_name), new_format=True).device_id
# If no device ID found, generate a new one and set the cookie
if device_id is None:
device_id = str(uuid.uuid4())
amp_cookie_value = AmplitudeCookie.generate(device_id)
# For Browser SDK 2.0: AmplitudeCookie.generate(device_id, new_format=True)
response.set_cookie(amp_cookie_name, amp_cookie_value,
domain='.your-domain.com', # this should be the same domain used by the Amplitude JS SDK
httponly=False,
secure=False
)
Was this helpful?