Skip to main content

Legacy Python Server SDK

Installation​

Install the sdk using pip3:

NOTE: The Statsig SDK is not compatible with python 2. You must be on python 3.7+ to use the Statsig SDK.

pip3 install statsig

Initialize the SDK​

After installation, you will need to initialize the SDK using a Server Secret Key from the statsig console.

info

Do NOT embed your Server Secret Key in client-side applications, or expose it in any external-facing documents. However, if you accidentally expose it, you can create a new one in the Statsig console.

There is also an optional parameter named options that allows you to pass in a StatsigOptions to customize the SDK.
from statsig import statsig

statsig.initialize("server-secret-key")

# or with StatsigOptions
options = StatsigOptions(tier=StatsigEnvironmentTier.development)
statsig.initialize("server-secret-key", options)

# check if sdk is initialized
initialized = statsig.is_initialized()
initialize will perform a network request. After initialize completes, virtually all SDK operations will be synchronous (See Evaluating Feature Gates in the Statsig SDK). The SDK will fetch updates from Statsig in the background, independently of your API calls.

Working with the SDK​

Checking a Feature Flag/Gate​

Now that your SDK is initialized, let's fetch a Feature Gate. Feature Gates can be used to create logic branches in code that can be rolled out to different users from the Statsig Console. Gates are always CLOSED or OFF (think return false;) by default.

From this point on, all APIs will require you to specify the user (see Statsig user) associated with the request. For example, check a gate for a certain user like this:

from statsig.statsig_user import StatsigUser

...

statsig.check_gate(StatsigUser("user-id"), "gate-name")

Reading a Dynamic Config​

Feature Gates can be very useful for simple on/off switches, with optional but advanced user targeting. However, if you want to be able send a different set of values (strings, numbers, and etc.) to your clients based on specific user attributes, e.g. country, Dynamic Configs can help you with that. The API is very similar to Feature Gates, but you get an entire json object you can configure on the server and you can fetch typed parameters from it. For example:

config = statsig.get_config(StatsigUser("user-id"), "config-name")

config_json = config.get_value()

Getting a Layer/Experiment​

Then we have Layers/Experiments, which you can use to run A/B/n experiments. We offer two APIs, but we recommend the use of layers to enable quicker iterations with parameter reuse.

# Values via getLayer

layer = statsig.get_layer(user, "user_promo_experiments")
title = layer.get("title", "Welcome to Statsig!")
discount = layer.get("discount", 0.1)

# or, via getExperiment

title_exp = statsig.get_experiment(user, "new_user_promo_title")
price_exp = statsig.get_experiment(user, "new_user_promo_price")

title = title_exp.get("title", "Welcome to Statsig!")
discount = price_exp.get("discount", 0.1)

...

price = msrp * (1 - discount)

Asynchronous APIs

We mentioned earlier that after calling initialize most SDK APIs would run synchronously, so why are getConfig and checkGate asynchronous?

The main reason is that older versions of the SDK might not know how to interpret new types of gate conditions. In such cases the SDK will make an asynchronous call to our servers to fetch the result of a check. This can be resolved by upgrading the SDK, and we will warn you if this happens.

For more details, read our blog post about SDK evaluations. If you have any questions, please ask them in our Feedback Repository.

Logging an Event​

Now that you have a Feature Gate or an Experiment set up, you may want to track some custom events and see how your new features or different experiment groups affect these events. This is super easy with Statsig - simply call the Log Event API and specify the user and event name to log; you additionally provide some value and/or an object of metadata to be logged together with the event:

from statsig.statsig_user import StatsigUser
from statsig.statsig_event import StatsigEvent

statsig.log_event(StatsigEvent(StatsigUser("user-id"), "event-name"))

Python supports retry_queue_size, which allows you to adjust the memory allocated for handling retries. While service outages are rare, increasing the retry_queue_size can help minimize event loss by providing additional memory to buffer events during such occurrences. This option is generally not needed for typical use but offers added flexibility in exceptional situations.

Learn more about identifying users, group analytics, and best practices for logging events in the logging events guide.

Retrieving Feature Gate Metadata​

In certain scenarios, you may need more information about a gate evaluation than just a boolean value. For additional metadata about the evaluation, use the Get Feature Gate API, which returns a FeatureGate object:

gate = statsig.get_feature_gate(StatsigUser("user-id"), "gate-name")
details = gate.get_evaluation_details()
if (details is not None and details.reason == "Unrecognized") {
  // gate was not defined in the payload that initialized the sdk
}

Statsig User​

When calling APIs that require a user, you should pass as much information as possible in order to take advantage of advanced gate and config conditions (like country or OS/browser level checks), and correctly measure impact of your experiments on your metrics/events. At least one identifier, either userID or a Custom ID, is required to provide a consistent experience for a given user (as explained here).

Besides userID, we also have email, ip, userAgent, country, locale and appVersion as top-level fields on StatsigUser. In addition, you can pass any key-value pairs in an object/dictionary to the custom field and be able to create targeting based on them.

Note that while typing is lenient on the StatsigUser object to allow you to pass in numbers, strings, arrays, objects, and potentially even enums or classes, the evaluation operators will only be able to operate on primitive types - mostly strings and numbers. While we attempt to smartly cast custom field types to match the operator, we cannot guarantee evaluation results for other types. For example, setting an array as a custom field will only ever be compared as a string - there is no operator to match a value in that array.

user = StatsigUser(
    user_id="123",
    email="testuser@statsig.com",
    ip="192.168.1.101",
    user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.78 Safari/537.36",
    country="US",
    locale="en_US",
    app_version="4.3.0",
    custom={"cohort": "June 2021"},
    private_attributes={"gender": "female"},
)

Private Attributes​

Have sensitive user PII data that should not be logged? No problem, we have a solution for it! On the StatsigUser object we also have a field called privateAttributes, which is a simple object/dictionary that you can use to set private user attributes. Any attribute set in privateAttributes will only be used for evaluation/targeting, and removed from any logs before they are sent to Statsig server.

For example, if you have feature gates that should only pass for users with emails ending in "@statsig.com", but do not want to log your users' email addresses to Statsig, you can simply add the key-value pair { email: "my_user@statsig.com" } to privateAttributes on the user and that's it!

Statsig Options​

statsig.initialize() takes an optional parameter options in addition to the secret key that you can provide to customize the Statsig client. Create a StatsigOptions class to pass in with the following available parameters:

(unit of measure for time related options is seconds)

  • tier: StatsigEnvironmentTier | str, default None

    • Sets the environment tier (for gates to evaluate differently in development and production)
    • You can set an environment tier with the StatsigEnvironmentTier enum or just as a str

  • timeout: timeout: int = None

    • Enforces a minimum timeout on network requests from the SDK

  • init_timeout: init_timeout: int = None

    • Sets the maximum timeout on download config specs and id lists network requests for initialization

  • rulesets_sync_interval: rulesets_sync_interval: int = 10

    • How often the SDK updates rulesets from Statsig servers

  • idlists_sync_interval: idlists_sync_interval: int = 60

    • How often the SDK updates idlists from Statsig servers

  • local_mode: local_mode: bool=False

    • Disables all network requests. SDK returns default values and will not log events. Useful in combination with overrides to mock behavior for tests

  • bootstrap_values: bootstrap_values: str = null

    • a string that represents all rules for all feature gates, dynamic configs and experiments. It can be provided to bootstrap the Statsig server SDK at initialization in case your server runs into network issue or Statsig server is down temporarily.

  • rules_updated_callback: rules_updated_callback: typing.Callable = None,

    • a callback function that's called whenever we have an update for the rules; it's called with a logical timestamp and a JSON string (used as is for bootstrapValues mentioned above). Note that as of v0.6.0, this will be called from a background thread that the SDK uses to update config values.

  • event_queue_size: event_queue_size: int = 500

    • The number of events to batch before flushing the queue to the network. Default 500.
    • Note that events are also batched every minute by a background thread

  • data_store: data_store: IDataStore = None

    • A data store with custom storage behavior for config specs. Can be used to bootstrap Statsig server (takes priority over bootstrap_values).

  • *proxy_configs: *proxy_configs: Optional[Dict[NetworkEndpoint, ProxyConfig]] = None,

    • Configuration network for each endpoint, for example, download_config_spec, get_id_lists

  • fallback_to_statsig_api: fallback_to_statsig_api: Optional[bool] = False,

    • Fallback to Statsig CDN for download config specs and get id lists if the overridden api failed.

  • initialize_sources: initialize_sources: Optional[List[DataSource]] = None,

    • List of sources SDK tries to get download_config_specs from when initialize. The list is ordered, SDK tries to get source from first element, and stops when getting dcs successfully

  • config_sync_sources: config_sync_sources: Optional[List[DataSource]] = None,

    • List of sources SDK tries to get download_config_specs from when downloading. The list is ordered, SDK tries to get source from first element, and stops when getting dcs successfully

Example:

from statsig import statsig, StatsigEnvironmentTier, StatsigOptions

options = StatsigOptions(None, StatsigEnvironmentTier.development)
statsig.initialize("secret-key", options).wait()

You can also use the set_environment_parameter function, but that takes in string values only:

from statsig import statsig, StatsigEnvironmentTier, StatsigOptions

options = StatsigOptions()
options.set_environment_parameter("tier", StatsigEnvironmentTier.development.value)
statsig.initialize("secret-key", options).wait()

Shutting Statsig Down​

Because we batch and periodically flush events, some events may not have been sent when your app/server shuts down.

To make sure all logged events are properly flushed, you should tell Statsig to shutdown when your app/server is closing:

statsig.shutdown()

Client SDK Bootstrapping | SSR v0.12.0+​

The Python server SDK, starting in 0.12.0 supports generating the initializeValues needed to bootstrap a Statsig Client SDK preventing a round trip to Statsig servers. This can also be used with web [ @statsig/js-client, @statsig/react-bindings] SDKs to perform server side rendering (SSR).

values = statsig.get_client_initialize_response(user); # dict() | None

# To apply local overrides, set include_local_overrides = True (python sdk v0.32.0+)
values = statsig.get_client_initialize_response(user=user, include_local_overrides=True); # dict() | None

Working with IP or UserAgent Values​

This will not automatically use the ip, or userAgent for gate evaluation as Statsig servers would, since there is no request from the client SDK specifying these values. If you want to use conditions like IP, or conditions which are inferred from the IP/UA like: Browser Name or Version, OS Name or Version, Country, you must manually set the ip and userAgent field on the user object when calling get_client_initialize_response.

Working with stableID​

There is no auto-generated stableID for device based experimentation, since the server generates the initialize response without any information from the client SDK. If you wish to run a device based experiment while using the server to generate the initialize response, we recommend you:

  1. Create a customID in the Statsig console. See experimenting on custom IDs for more information.
  2. Generate an ID on the server, and set it in a cookie to be used on the client side as well.
  3. Set that ID as the customID on the StatsigUser object when generating the initialize response from the SDK.
  4. Get that ID from the cookie, and set it as the customID on the StatsigUser object when using the client SDK, so all event data and exposure checks tie back to the same user.

Alternatively, if you wish to use the stableID field rather than a custom ID, you still need to do step (2) above. Then:

  • Override the stableID in the client SDK by getting the value from the cookie and setting the overrideStableID parameter in StatsigOptions
  • Set the stableID field on the StatsigUser object in the customIDs map when generating the initialize response from the SDK

getClientInitializeResponse and the legacy JS SDK​

If you are migrating from the legacy JS Client, you will need to make some updates to how your server SDK generates values. The default hashing algorithm was changed from sha256 to djb2 for performance and size reasons.

See the Bootstrapping Migration Docs for more.

Multiple Statsig SDK Instances​

Our documentation up to this point guides you through setting up your Statsig integration via the singleton Statsig.

There are cases where you may need to create multiple instances of the Statsig SDK. Each SDK supports this as well - the Statsig singleton wraps a single instance of the SDK that you can instantiate. NOTE: currently all sdk instances will use the same keys when interacting with a Data Store/Data Adapter. You will not be able to isolate multiple instances of the sdk in your data store.

All top level static methods from the singleton carry over as instance methods. To create an instance of the Statsig sdk:

sdk_instance = StatsigServer()
sdk_instance.initialize(secret_key, options);

SDK Monitoring v0.49.0+​

The SDK provide an option to integrate with your preferred observability tool to monitor the SDK's behavior and performance. For detailed information and metrics emitted, please see sdk monitoring

ObservabilityClient interface​

The SDK provides the following interface methods to track various metrics:

class ObservabilityClient:
    """
    An interface for observability clients that allows users to plug in their
    own observability integration for metrics collection and monitoring.
    """

    def init(self, *args, **kwargs):
        """
        Initializes the observability client with necessary configuration.

        :param args: Positional arguments for initialization.
        :param kwargs: Keyword arguments for initialization.
        """

    def increment(self, metric_name: str, value: int = 1, tags: Optional[Dict[str, Any]] = None) -> None:
        """
        Increment a counter metric.

        :param metric_name: The name of the metric to increment.
        :param value: The value by which the counter should be incremented (default is 1).
        :param tags: Optional dictionary of tags for metric dimensions.
        """

    def gauge(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None) -> None:
        """
        Set a gauge metric.

        :param metric_name: The name of the metric to set.
        :param value: The value to set the gauge to.
        :param tags: Optional dictionary of tags for metric dimensions.
        """

    def distribution(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None) -> None:
        """
        Record a distribution metric for tracking statistical data.

        :param metric_name: The name of the metric to record.
        :param value: The recorded value for the distribution metric.
        :param tags: Optional dictionary of tags that represent dimensions to associate with the metric.
        """

    def should_enable_high_cardinality_for_this_tag(self, tag: str) -> bool:
        """
        Determine if a high cardinality tag should be logged. See the list of high cardinality tags https://docs.statsig.com/sdk_monitoring/

        :param tag: The tag to check for high cardinality enabled.
        """
        return False

    def shutdown(self) -> None:
        """
        Shutdown the observability client.
        """

Setup​

Example of setting up metrics collection with StatsD

class ExampleObservationClient(ObservabilityClient):
    def init(self):
        options = {
            "your options here"
        }
        initialize(**options)

    def increment(self, metric_name: str, value: int = 1, tags: Optional[Dict[str, Any]] = None) -> None:
        formatted_tags = format_tags(tags)
        statsd.increment(metric_name, value, tags=formatted_tags)

    def gauge(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None) -> None:
        formatted_tags = format_tags(tags)
        statsd.gauge(metric_name, value, tags=formatted_tags)

    def distribution(self, metric_name: str, value: float, tags: Optional[Dict[str, Any]] = None) -> None:
        formatted_tags = format_tags(tags)
        statsd.distribution(metric_name, value, tags=formatted_tags)

    def should_enable_high_cardinality_for_this_tag(self, tag: str) -> bool:
        return True

    def shutdown(self) -> None:
        pass

def format_tags(tags: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
    if not tags:
        return None
    return [f"{key}:{value}" for key, value in tags.items()]

options = StatsigOptions(
    observability_client=ExampleObservationClient(),
)
statsig.initialize(secret_key, options)

Proxy Configurations v0.39.0+​

Advance SDK Network setup, provide options to configure network protocol and proxy address for individual network endpoint. This option is provided for better integration with Statsig Forward Proxy

Authentication v0.45.0+​

With more advanced requirements on security, we also support TLS and mTLS for grpc streaming (protocol=GRPC_WEBSOCKET) You need to configure ProxyConfig for each endpoint

Authentication

authentication_mode: TLS, mTLS, disabled // default is disabled
tls_client_cert_path // File path to client certification 
tls_client_key_path // File path to client key password
tls_ca_cert_path  // File path to root TLS certificate

Initialization

To setup grpc streaming for download config spec. And use default behavior for get_id_lists and log_events endpoints.

Basic setup to stream download config spec from forward proxy

  proxyAddress = "0.0.0.0:50051" // local address update to your address
  Statsig.initialize(secret_key, StatsigOptions(proxy_configs={
            NetworkEndpoint.DOWNLOAD_CONFIG_SPECS: ProxyConfig(NetworkProtocol.GRPC_WEBSOCKET, proxyAddress)}))

When the SDK is disconnected from forward proxy when use grpc_websocket, the sdk will retry connection with exponential backoff, after push_worker_failover_threshold retries, the sdk will start polling from Statsig until reconnecting to the forward proxy. You can customize Streaming Failover Behavior. You can also define the sources/endpoints SDK poll from, SDK will try from source at index 0, and stops trying if get a response.


statsigOptions = StatsigOptions(
    proxy_configs={
        NetworkEndpoint.DOWNLOAD_CONFIG_SPECS: ProxyConfig(
            protocol=NetworkProtocol.GRPC_WEBSOCKET,
            proxy_address=address,
            push_worker_failover_threshold=1, # start polling from Statsig endpoint after 1 retry failed
            # 1st retry 5000 ms later, 2nd retry 2 * 5000ms = 10 seconds ....
            retry_backoff_multiplier=2, 
            max_retry_attempt=8,
            retry_backoff_base_ms=5000
        )
    },
    # Get from network first, which is forward proxy here, if fails, try datastore, if fails try poll from Statsig endpoint
    initialize_sources=[
        DataSource.NETWORK,
        DataSource.DATASTORE,
        DataSource.STATSIG_NETWORK,
    ],
)

Local Overrides ​

If you want to locally override gates/configs/experiments/layers, there are a set of override APIs as follows. Coupling this with StatsigOptions.localMode can be useful when writing unit tests.

# Adding/Removing gate overrides
statsig.override_gate("a_gate_name", true, "a_user_id")
statsig.remove_gate_override("a_gate_name", "a_user_id")

# Adding/Removing config overrides
statsig.override_config("a_config_name", {"key": "value"}, "a_user_id")
statsig.remove_config_override("a_config_name", "a_user_id")

# Adding/Removing experiment overrides
statsig.override_experiment("an_experiment_name", {"key": "value"}, "a_user_id")
statsig.remove_experiment_override("an_experiment_name", "a_user_id")

# Remove All Overrides
statsig.remove_all_overrides()

# You can also override with custom ids
custom_id_user = StatsigUser("a_user_id", custom_ids={"statsigId": "a_statsig_id"})
statsig.override_gate("a_gate_name", true, "a_statsig_id")

# Local overrides will prioritize override with userId, then look up the custom id to override. 
# To prevent clashing overrides, it is recommended to not use the same value for userId and customIds for different users.
note
  1. These only apply locally - they do not update definitions in the Statsig console or elsewhere.
  2. The local override API is not designed to be a full mock. They are only a convenient way to override the value of the gate/config/etc.

FAQ​

How do I run experiments for logged out users?​

See the guide on device level experiments

How can I mock Statsig for testing?​

The python server SDK, starting in version 0.5.1+, supports a few features to make testing easier.

First, there is a StatsigOption parameter called localMode. Setting localMode to true will cause the SDK to never hit the network, and only return default values. This is perfect for dummy environments or test environments that should not access the network.

Next, there are the overrideGate and overrideConfig APIs on the global statsig interface, see Local Overrides

These can be used to set a gate or config override for a specific user, or for all users (by not providing a specific user ID).

We suggest you enable localMode and then override gates/configs/experiments to specific values to test the various code flows you are building.

Can I generate the initialize response for a client SDK using the Python server SDK?​

Yes. See Client Initialize Response.

Reference​

StatsigUser​

@dataclass
class StatsigUser:
    """An object of properties relating to the current user
    user_id or at least a custom ID is required: learn more https://docs.statsig.com/concepts/user#why-is-an-id-always-required-for-server-sdks
    Provide as many as possible to take advantage of advanced conditions in the statsig console
    A dictionary of additional fields can be provided under the custom field
    Set private_attributes for any user property you need for gate evaluation but prefer stripped from logs/metrics
    """
    user_id: Optional[str] = None
    email: Optional[str] = None
    ip: Optional[str] = None
    user_agent: Optional[str] = None
    country: Optional[str] = None
    locale: Optional[str] = None
    app_version: Optional[str] = None 
    custom: Optional[dict] = None # key: string, value: string
    private_attributes: Optional[dict] = None # key: string, value: string
    custom_ids: Optional[dict] = None # key: string, value: string

StatsigOptions​

class StatsigOptions:
    """An object of properties for initializing the sdk with additional parameters"""

    def __init__(
        self,
        api: Optional[str] = None,
        api_for_download_config_specs: Optional[str] = None,
        api_for_get_id_lists: Optional[str] = None,
        api_for_log_event: Optional[str] = None,
        tier: Union[str, StatsigEnvironmentTier, None] = None,
        init_timeout: Optional[int] = None,
        timeout: Optional[int] = None,
        rulesets_sync_interval: int = DEFAULT_RULESET_SYNC_INTERVAL,
        idlists_sync_interval: int = DEFAULT_IDLIST_SYNC_INTERVAL,
        local_mode: bool = False,
        bootstrap_values: Optional[str] = None,
        rules_updated_callback: Optional[Callable] = None,
        event_queue_size: Optional[int] = DEFAULT_EVENT_QUEUE_SIZE,
        data_store: Optional[IDataStore] = None,
        idlists_thread_limit: int = DEFAULT_IDLISTS_THREAD_LIMIT,
        logging_interval: int = DEFAULT_LOGGING_INTERVAL, #deprecated
        disable_diagnostics: bool = False,
        custom_logger: Optional[OutputLogger] = None,
        enable_debug_logs = False,
        disable_all_logging = False,
        evaluation_callback: Optional[Callable[[Union[Layer, DynamicConfig, FeatureGate]], None]] = None,
        retry_queue_size: int = DEFAULT_RETRY_QUEUE_SIZE,
        proxy_configs: Optional[Dict[NetworkEndpoint, ProxyConfig]] = None,
        fallback_to_statsig_api: Optional[bool] = False,
        initialize_sources: Optional[List[DataSource]] = None,
        config_sync_sources: Optional[List[DataSource]] = None,
    ):

FeatureGate​

class FeatureGate:

    def get_value(self):
        """Returns the underlying value of this FeatureGate"""

    def get_name(self):
        """Returns the name of this FeatureGate"""

    def get_evaluation_details(self):
        """Returns the evaluation detail of this FeatureGate"""

DynamicConfig​

class DynamicConfig:
    def get(self, key, default=None):
        """Returns the value of the config at the given key
        or the provided default if the key is not found
        """

    def get_typed(self, key, default=None):
        """Returns the value of the config at the given key
        iff the type matches the type of the provided default.
        Otherwise, returns the default value
        """

    def get_value(self):
        """Returns the underlying value of this DynamicConfig"""

    def get_name(self):
        """Returns the name of this DynamicConfig"""

    def get_evaluation_details(self):
        """Returns the evaluation detail of this DynamicConfig"""

Layer​

class Layer:
    def get(self, key, default=None):
        """Returns the value of the layer at the given key
        or the provided default if the key is not found
        """

    def get_typed(self, key, default=None):
        """Returns the value of the layer at the given key
        iff the type matches the type of the provided default.
        Otherwise, returns the default value
        """

    def get_name(self):
        """Returns the name of this Layer"""

    def get_values(self):
        """Returns all the values in this Layer but does not trigger an exposure log"""

    def get_evaluation_details(self):
        """Returns the evaluation detail of this Layer"""

EvaluationDetails​

class EvaluationDetails:
    reason: EvaluationReason
    config_sync_time: int
    init_time: int
    server_time: int

class EvaluationReason(str, Enum):
    network = "Network"
    local_override = "LocalOverride"
    unrecognized = "Unrecognized"
    uninitialized = "Uninitialized"
    bootstrap = "Bootstrap"
    data_adapter = "DataAdapter"
    unsupported = "Unsupported"
    error = "error"

DataStore​

class IDataStore:
    def get(self, key: str) -> Optional[str]:
        return None

    def set(self, key: str, value: str):
        pass

    def shutdown(self):
        pass

    def should_be_used_for_querying_updates(self, key: str) -> bool:
        return False

ForwardProxy - ProxyConfig​

class NetworkProtocol(Enum):
    HTTP = "http"
    GRPC = "grpc"
    GRPC_WEBSOCKET = "grpc_websocket"


class NetworkEndpoint(Enum):
    LOG_EVENT = "log_event"
    DOWNLOAD_CONFIG_SPECS = "download_config_specs"
    GET_ID_LISTS = "get_id_lists"
    ALL = "all"

class ProxyConfig:
    def __init__(
        self,
        protocol: NetworkProtocol,
        proxy_address: str,
        # Websocket worker failover config
        max_retry_attempt: Optional[int] = None, # default is 10
        retry_backoff_multiplier: Optional[int] = None, # default is # default is 5
        retry_backoff_base_ms: Optional[int] = None, # default is 10,000 ms
        # Push worker failback to polling threshold, fallback immediate set 0,
        # n means fallback after n retry failed
        push_worker_failover_threshold: Optional[int] = None, # default is 4, about 30 minutes
    ):
        self.proxy_address = proxy_address
        self.protocol = protocol
        self.max_retry_attempt = max_retry_attempt
        self.retry_backoff_multiplier = retry_backoff_multiplier
        self.retry_backoff_base_ms = retry_backoff_base_ms
        self.push_worker_failover_threshold = push_worker_failover_threshold