Legacy Python Server SDK

from statsig.statsig_user import StatsigUser

...

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

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

config_json = config.get_value()

# 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)

gate = statsig.get_feature_gate(StatsigUser("user-id"), "gate-name")
print(gate.name)  # 'gate-name'
print(gate.value)  # True or False
print(gate.rule_id)  # rule ID that was evaluated
print(gate.evaluation_details)  # evaluation metadata

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

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

from statsig import statsig, StatsigEnvironmentTier, StatsigOptions

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

from statsig import statsig, StatsigEnvironmentTier, StatsigOptions

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

statsig.shutdown()

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

# 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.

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

  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)}))

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,
    ],
)

@dataclass
class StatsigUser:
    """An object of properties relating to the current user
    user_id or customID is required: /sdks/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

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,
    ):

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"""

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"""

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"""

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"

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

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