""" Class encapsulating all SDK functionality. """

def __init__(

self,

datafile: Optional[str] = None,

event_dispatcher: Optional[CustomEventDispatcher] = None,

logger: Optional[_logging.Logger] = None,

error_handler: Optional[BaseErrorHandler] = None,

skip_json_validation: Optional[bool] = False,

user_profile_service: Optional[UserProfileService] = None,

sdk_key: Optional[str] = None,

config_manager: Optional[BaseConfigManager] = None,

notification_center: Optional[NotificationCenter] = None,

event_processor: Optional[BaseEventProcessor] = None,

datafile_access_token: Optional[str] = None,

default_decide_options: Optional[list[str]] = None,

event_processor_options: Optional[dict[str, Any]] = None,

settings: Optional[OptimizelySdkSettings] = None,

cmab_service: Optional[DefaultCmabService] = None,

) -> None:

""" Optimizely init method for managing Custom projects.

self.logger_name = '.'.join([__name__, self.__class__.__name__])

self.is_valid = True

self.event_dispatcher = event_dispatcher or EventDispatcher

self.logger = _logging.adapt_logger(logger or _logging.NoOpLogger())

self.error_handler = error_handler or NoOpErrorHandler

self.config_manager: BaseConfigManager = config_manager # type: ignore[assignment]

self.notification_center = notification_center or NotificationCenter(self.logger)

event_processor_defaults = {

'batch_size': 1,

'flush_interval': 30,

'timeout_interval': 5,

'start_on_init': True

}

if event_processor_options:

event_processor_defaults.update(event_processor_options)

self.event_processor = event_processor or BatchEventProcessor(

self.event_dispatcher,

logger=self.logger,

notification_center=self.notification_center,

**event_processor_defaults # type: ignore[arg-type]

)

self.default_decide_options: list[str]

if default_decide_options is None:

self.default_decide_options = []

else:

self.default_decide_options = default_decide_options

if isinstance(self.default_decide_options, list):

self.default_decide_options = self.default_decide_options[:]

else:

self.logger.debug('Provided default decide options is not a list.')

self.default_decide_options = []

self.sdk_settings: OptimizelySdkSettings = settings # type: ignore[assignment]

try:

self._validate_instantiation_options()

except exceptions.InvalidInputException as error:

self.is_valid = False

# We actually want to log this error to stderr, so make sure the logger

# has a handler capable of doing that.

self.logger = _logging.reset_logger(self.logger_name)

self.logger.exception(str(error))

return

config_manager_options: dict[str, Any] = {

'datafile': datafile,

'logger': self.logger,

'error_handler': self.error_handler,

'notification_center': self.notification_center,

'skip_json_validation': skip_json_validation,

}

if not self.config_manager:

if sdk_key:

config_manager_options['sdk_key'] = sdk_key

if datafile_access_token:

config_manager_options['datafile_access_token'] = datafile_access_token

self.config_manager = AuthDatafilePollingConfigManager(**config_manager_options)

else:

self.config_manager = PollingConfigManager(**config_manager_options)

else:

self.config_manager = StaticConfigManager(**config_manager_options)

self.odp_manager: OdpManager

self._setup_odp(self.config_manager.get_sdk_key())

self.event_builder = event_builder.EventBuilder()

# Initialize CMAB components

if cmab_service:

self.cmab_service = cmab_service

else:

# Get custom prediction endpoint from settings if provided

cmab_prediction_endpoint = None

if self.sdk_settings and self.sdk_settings.cmab_prediction_endpoint:

cmab_prediction_endpoint = self.sdk_settings.cmab_prediction_endpoint

self.cmab_client = DefaultCmabClient(

retry_config=CmabRetryConfig(),

logger=self.logger,

prediction_endpoint=cmab_prediction_endpoint

)

self.cmab_cache: LRUCache[str, CmabCacheValue] = LRUCache(DEFAULT_CMAB_CACHE_SIZE,

DEFAULT_CMAB_CACHE_TIMEOUT)

self.cmab_service = DefaultCmabService(

cmab_cache=self.cmab_cache,

cmab_client=self.cmab_client,

logger=self.logger

)

self.decision_service = decision_service.DecisionService(self.logger, user_profile_service, self.cmab_service)

self.user_profile_service = user_profile_service

def _get_variation_key(self, variation: Optional[Union[entities.Variation, VariationDict]]) -> Optional[str]:

"""Helper to extract variation key from either dict (holdout) or Variation object.

if variation is None:

return None

try:

# Try dict access first (for holdouts)

if isinstance(variation, dict):

return variation.get('key')

# Otherwise assume it's a Variation entity object

else:

return variation.key

except (AttributeError, KeyError, TypeError):

self.logger.warning(f"Unable to extract variation key from {type(variation)}")

return None

def _get_variation_id(self, variation: Optional[Union[entities.Variation, VariationDict]]) -> Optional[str]:

"""Helper to extract variation id from either dict (holdout) or Variation object.

if variation is None:

return None

try:

# Try dict access first (for holdouts)

if isinstance(variation, dict):

return variation.get('id')

# Otherwise assume it's a Variation entity object

else:

return variation.id

except (AttributeError, KeyError, TypeError):

self.logger.warning(f"Unable to extract variation id from {type(variation)}")

return None

def _get_feature_enabled(self, variation: Optional[Union[entities.Variation, VariationDict]]) -> bool:

"""Helper to extract featureEnabled flag from either dict (holdout) or Variation object.

if variation is None:

return False

try:

if isinstance(variation, dict):

feature_enabled = variation.get('featureEnabled', False)

return bool(feature_enabled) if feature_enabled is not None else False

else:

return variation.featureEnabled if hasattr(variation, 'featureEnabled') else False

except (AttributeError, KeyError, TypeError):

return False

def _validate_instantiation_options(self) -> None:

""" Helper method to validate all instantiation parameters.

if self.config_manager and not validator.is_config_manager_valid(self.config_manager):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('config_manager'))

if not validator.is_event_dispatcher_valid(self.event_dispatcher):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('event_dispatcher'))

if not validator.is_logger_valid(self.logger):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('logger'))

if not validator.is_error_handler_valid(self.error_handler):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('error_handler'))

if not validator.is_notification_center_valid(self.notification_center):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('notification_center'))

if not validator.is_event_processor_valid(self.event_processor):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('event_processor'))

if not isinstance(self.sdk_settings, OptimizelySdkSettings):

if self.sdk_settings is not None:

self.logger.debug('Provided sdk_settings is not an OptimizelySdkSettings instance.')

self.sdk_settings = OptimizelySdkSettings()

if self.sdk_settings.segments_cache:

if not validator.is_segments_cache_valid(self.sdk_settings.segments_cache):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('segments_cache'))

if self.sdk_settings.odp_segment_manager:

if not validator.is_segment_manager_valid(self.sdk_settings.odp_segment_manager):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('segment_manager'))

if self.sdk_settings.odp_event_manager:

if not validator.is_event_manager_valid(self.sdk_settings.odp_event_manager):

raise exceptions.InvalidInputException(enums.Errors.INVALID_INPUT.format('event_manager'))

def _validate_user_inputs(

self, attributes: Optional[UserAttributes] = None, event_tags: Optional[EventTags] = None

) -> bool:

""" Helper method to validate user inputs.

if attributes and not validator.are_attributes_valid(attributes):

self.logger.error('Provided attributes are in an invalid format.')

self.error_handler.handle_error(exceptions.InvalidAttributeException(enums.Errors.INVALID_ATTRIBUTE_FORMAT))

return False

if event_tags and not validator.are_event_tags_valid(event_tags):

self.logger.error('Provided event tags are in an invalid format.')

self.error_handler.handle_error(exceptions.InvalidEventTagException(enums.Errors.INVALID_EVENT_TAG_FORMAT))

return False

return True

def _send_impression_event(

self, project_config: project_config.ProjectConfig,

experiment: Optional[Union[entities.Experiment, entities.Holdout]],

variation: Optional[Union[entities.Variation, VariationDict]], flag_key: str, rule_key: str, rule_type: str,

enabled: bool, user_id: str, attributes: Optional[UserAttributes], cmab_uuid: Optional[str] = None

) -> None:

""" Helper method to send impression event.

if not experiment:

experiment = entities.Experiment.get_default()

variation_id = self._get_variation_id(variation) if variation is not None else None

user_event = user_event_factory.UserEventFactory.create_impression_event(

project_config, experiment, variation_id,

flag_key, rule_key, rule_type,

enabled, user_id, attributes, cmab_uuid

)

if user_event is None:

self.logger.error('Cannot process None event.')

return

self.event_processor.process(user_event)

# Kept for backward compatibility.

# This notification is deprecated and new Decision notifications

# are sent via their respective method calls.

if len(self.notification_center.notification_listeners[enums.NotificationTypes.ACTIVATE]) > 0:

log_event = event_factory.EventFactory.create_log_event(user_event, self.logger)

self.notification_center.send_notifications(

enums.NotificationTypes.ACTIVATE, experiment, user_id, attributes, variation, log_event.__dict__,

)

def _get_feature_variable_for_type(

self, project_config: project_config.ProjectConfig, feature_key: str, variable_key: str,

variable_type: Optional[str], user_id: str, attributes: Optional[UserAttributes]

) -> Any:

""" Helper method to determine value for a certain variable attached to a feature flag based on

if not validator.is_non_empty_string(feature_key):

self.logger.error(enums.Errors.INVALID_INPUT.format('feature_key'))

return None

if not validator.is_non_empty_string(variable_key):

self.logger.error(enums.Errors.INVALID_INPUT.format('variable_key'))

return None

if not isinstance(user_id, str):

self.logger.error(enums.Errors.INVALID_INPUT.format('user_id'))

return None

if not self._validate_user_inputs(attributes):

return None

feature_flag = project_config.get_feature_from_key(feature_key)

if not feature_flag:

return None

variable = project_config.get_variable_for_feature(feature_key, variable_key)

if not variable:

return None

# For non-typed method, use type of variable; else, return None if type differs

variable_type = variable_type or variable.type

if variable.type != variable_type:

self.logger.warning(

f'Requested variable type "{variable_type}", but variable is of '

f'type "{variable.type}". Use correct API to retrieve value. Returning None.'

)

return None

feature_enabled = False

source_info = {}

variable_value = variable.defaultValue

user_context = OptimizelyUserContext(self, self.logger, user_id, attributes, False)

decision_result = self.decision_service.get_variation_for_feature(project_config, feature_flag, user_context)

decision = decision_result['decision']

if decision.variation:

feature_enabled = self._get_feature_enabled(decision.variation)

if feature_enabled:

variable_value = project_config.get_variable_value_for_variation(variable, decision.variation)

self.logger.info(

f'Got variable value "{variable_value}" for '

f'variable "{variable_key}" of feature flag "{feature_key}".'

)

else:

self.logger.info(

f'Feature "{feature_key}" is not enabled for user "{user_id}". '

f'Returning the default variable value "{variable_value}".'

)

else:

self.logger.info(

f'User "{user_id}" is not in any variation or rollout rule. '

f'Returning default value for variable "{variable_key}" of feature flag "{feature_key}".'

)

if decision.source in (enums.DecisionSources.FEATURE_TEST, enums.DecisionSources.HOLDOUT):

source_info = {

'experiment_key': decision.experiment.key if decision.experiment else None,

'variation_key': self._get_variation_key(decision.variation),

}

try:

actual_value = project_config.get_typecast_value(variable_value, variable_type)

except:

self.logger.error('Unable to cast value. Returning None.')

actual_value = None

self.notification_center.send_notifications(

enums.NotificationTypes.DECISION,

enums.DecisionNotificationTypes.FEATURE_VARIABLE,

user_id,

attributes or {},

{

'feature_key': feature_key,

'feature_enabled': feature_enabled,

'source': decision.source,

'variable_key': variable_key,

'variable_value': actual_value,

'variable_type': variable_type,

'source_info': source_info,

},

)

return actual_value

def _get_all_feature_variables_for_type(

self, project_config: project_config.ProjectConfig, feature_key: str,

user_id: str, attributes: Optional[UserAttributes],

) -> Optional[dict[str, Any]]:

""" Helper method to determine value for all variables attached to a feature flag.

if not validator.is_non_empty_string(feature_key):

self.logger.error(enums.Errors.INVALID_INPUT.format('feature_key'))

return None

if not isinstance(user_id, str):

self.logger.error(enums.Errors.INVALID_INPUT.format('user_id'))

return None

if not self._validate_user_inputs(attributes):

return None

feature_flag = project_config.get_feature_from_key(feature_key)

if not feature_flag:

return None

feature_enabled = False

source_info = {}

user_context = OptimizelyUserContext(self, self.logger, user_id, attributes, False)

decision = self.decision_service.get_variation_for_feature(project_config,

feature_flag,

user_context)['decision']

if decision.variation:

feature_enabled = self._get_feature_enabled(decision.variation)

if feature_enabled:

self.logger.info(

f'Feature "{feature_key}" is enabled for user "{user_id}".'

)

else:

self.logger.info(

f'Feature "{feature_key}" is not enabled for user "{user_id}".'

)

else:

self.logger.info(

f'User "{user_id}" is not in any variation or rollout rule. '

f'Returning default value for all variables of feature flag "{feature_key}".'

)

all_variables = {}

for variable_key, variable in feature_flag.variables.items():

variable_value = variable.defaultValue

if feature_enabled:

variable_value = project_config.get_variable_value_for_variation(variable, decision.variation)

self.logger.debug(

f'Got variable value "{variable_value}" for '

f'variable "{variable_key}" of feature flag "{feature_key}".'

)

try:

actual_value = project_config.get_typecast_value(variable_value, variable.type)

except:

self.logger.error('Unable to cast value. Returning None.')

actual_value = None

all_variables[variable_key] = actual_value

if decision.source == enums.DecisionSources.FEATURE_TEST:

source_info = {

'experiment_key': decision.experiment.key if decision.experiment else None,

'variation_key': self._get_variation_key(decision.variation),

}

self.notification_center.send_notifications(

enums.NotificationTypes.DECISION,

enums.DecisionNotificationTypes.ALL_FEATURE_VARIABLES,

user_id,

attributes or {},

{

'feature_key': feature_key,

'feature_enabled': feature_enabled,

'variable_values': all_variables,

'source': decision.source,

'source_info': source_info,

},

)

return all_variables

def activate(self, experiment_key: str, user_id: str, attributes: Optional[UserAttributes] = None) -> Optional[str]:

""" Buckets visitor and sends impression event to Optimizely.

if not self.is_valid:

self.logger.error(enums.Errors.INVALID_OPTIMIZELY.format('activate'))

return None

if not validator.is_non_empty_string(experiment_key):

self.logger.error(enums.Errors.INVALID_INPUT.format('experiment_key'))

return None

if not isinstance(user_id, str):

self.logger.error(enums.Errors.INVALID_INPUT.format('user_id'))

return None

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('activate'))

return None

variation_key = self.get_variation(experiment_key, user_id, attributes)

if not variation_key:

self.logger.info(f'Not activating user "{user_id}".')

return None

experiment = project_config.get_experiment_from_key(experiment_key)

variation = project_config.get_variation_from_key(experiment_key, variation_key)

if not variation or not experiment:

self.logger.info(f'Not activating user "{user_id}".')

return None

# Create and dispatch impression event

self.logger.info(f'Activating user "{user_id}" in experiment "{experiment.key}".')

self._send_impression_event(project_config, experiment, variation, '', experiment.key,

enums.DecisionSources.EXPERIMENT, True, user_id, attributes)

# Handle both Variation entity and VariationDict

return variation['key'] if isinstance(variation, dict) else variation.key

def track(

self, event_key: str, user_id: str,

attributes: Optional[UserAttributes] = None,

event_tags: Optional[EventTags] = None

) -> None:

""" Send conversion event to Optimizely.

if not self.is_valid:

self.logger.error(enums.Errors.INVALID_OPTIMIZELY.format('track'))

return

if not validator.is_non_empty_string(event_key):

self.logger.error(enums.Errors.INVALID_INPUT.format('event_key'))

return

if not isinstance(user_id, str):

self.logger.error(enums.Errors.INVALID_INPUT.format('user_id'))

return

if not self._validate_user_inputs(attributes, event_tags):

return

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('track'))

return

event = project_config.get_event(event_key)

if not event:

self.logger.info(f'Not tracking user "{user_id}" for event "{event_key}".')

return

user_event = user_event_factory.UserEventFactory.create_conversion_event(

project_config, event_key, user_id, attributes, event_tags

)

if user_event is None:

self.logger.error('Cannot process None event.')

return

self.event_processor.process(user_event)

self.logger.info(f'Tracking event "{event_key}" for user "{user_id}".')

if len(self.notification_center.notification_listeners[enums.NotificationTypes.TRACK]) > 0:

log_event = event_factory.EventFactory.create_log_event(user_event, self.logger)

self.notification_center.send_notifications(

enums.NotificationTypes.TRACK, event_key, user_id, attributes, event_tags, log_event.__dict__,

)

def get_variation(

self, experiment_key: str, user_id: str, attributes: Optional[UserAttributes] = None

) -> Optional[str]:

""" Gets variation where user will be bucketed.

if not self.is_valid:

self.logger.error(enums.Errors.INVALID_OPTIMIZELY.format('get_variation'))

return None

if not validator.is_non_empty_string(experiment_key):

self.logger.error(enums.Errors.INVALID_INPUT.format('experiment_key'))

return None

if not isinstance(user_id, str):

self.logger.error(enums.Errors.INVALID_INPUT.format('user_id'))

return None

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('get_variation'))

return None

experiment = project_config.get_experiment_from_key(experiment_key)

variation_key = None

if not experiment:

self.logger.info(f'Experiment key "{experiment_key}" is invalid. Not activating user "{user_id}".')

return None

if not self._validate_user_inputs(attributes):

return None

user_context = OptimizelyUserContext(self, self.logger, user_id, attributes, False)

user_profile_tracker = user_profile.UserProfileTracker(user_id, self.user_profile_service, self.logger)

user_profile_tracker.load_user_profile()

variation_result = self.decision_service.get_variation(project_config, experiment,

user_context, user_profile_tracker)

variation = variation_result['variation']

user_profile_tracker.save_user_profile()

if variation:

variation_key = self._get_variation_key(variation)

if project_config.is_feature_experiment(experiment.id):

decision_notification_type = enums.DecisionNotificationTypes.FEATURE_TEST

else:

decision_notification_type = enums.DecisionNotificationTypes.AB_TEST

self.notification_center.send_notifications(

enums.NotificationTypes.DECISION,

decision_notification_type,

user_id,

attributes or {},

{'experiment_key': experiment_key, 'variation_key': variation_key},

)

return variation_key

def is_feature_enabled(self, feature_key: str, user_id: str, attributes: Optional[UserAttributes] = None) -> bool:

""" Returns true if the feature is enabled for the given user.

if not self.is_valid:

self.logger.error(enums.Errors.INVALID_OPTIMIZELY.format('is_feature_enabled'))

return False

if not validator.is_non_empty_string(feature_key):

self.logger.error(enums.Errors.INVALID_INPUT.format('feature_key'))

return False

if not isinstance(user_id, str):

self.logger.error(enums.Errors.INVALID_INPUT.format('user_id'))

return False

if not self._validate_user_inputs(attributes):

return False

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('is_feature_enabled'))

return False

feature = project_config.get_feature_from_key(feature_key)

if not feature:

return False

feature_enabled = False

source_info = {}

user_context = OptimizelyUserContext(self, self.logger, user_id, attributes, False)

decision = self.decision_service.get_variation_for_feature(project_config, feature, user_context)['decision']

cmab_uuid = decision.cmab_uuid

is_source_experiment = decision.source == enums.DecisionSources.FEATURE_TEST

is_source_rollout = decision.source == enums.DecisionSources.ROLLOUT

if decision.variation:

if self._get_feature_enabled(decision.variation) is True:

feature_enabled = True

if (is_source_rollout or not decision.variation) and project_config.get_send_flag_decisions_value():

self._send_impression_event(

project_config, decision.experiment, decision.variation, feature.key, decision.experiment.key if

decision.experiment else '', str(decision.source), feature_enabled, user_id, attributes, cmab_uuid

)

# Send event if Decision came from an experiment.

if is_source_experiment and decision.variation and decision.experiment:

source_info = {

'experiment_key': decision.experiment.key,

'variation_key': self._get_variation_key(decision.variation),

}

self._send_impression_event(

project_config, decision.experiment, decision.variation, feature.key, decision.experiment.key,

str(decision.source), feature_enabled, user_id, attributes, cmab_uuid

)

if feature_enabled:

self.logger.info(f'Feature "{feature_key}" is enabled for user "{user_id}".')

else:

self.logger.info(f'Feature "{feature_key}" is not enabled for user "{user_id}".')

self.notification_center.send_notifications(

enums.NotificationTypes.DECISION,

enums.DecisionNotificationTypes.FEATURE,

user_id,

attributes or {},

{

'feature_key': feature_key,

'feature_enabled': feature_enabled,

'source': decision.source,

'source_info': source_info,

},

)

return feature_enabled

def get_enabled_features(self, user_id: str, attributes: Optional[UserAttributes] = None) -> list[str]:

""" Returns the list of features that are enabled for the user.

enabled_features: list[str] = []

if not self.is_valid:

self.logger.error(enums.Errors.INVALID_OPTIMIZELY.format('get_enabled_features'))

return enabled_features

if not isinstance(user_id, str):

self.logger.error(enums.Errors.INVALID_INPUT.format('user_id'))

return enabled_features

if not self._validate_user_inputs(attributes):

return enabled_features

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('get_enabled_features'))

return enabled_features

for feature in project_config.feature_key_map.values():

if self.is_feature_enabled(feature.key, user_id, attributes):

enabled_features.append(feature.key)

return enabled_features

def get_feature_variable(

self, feature_key: str, variable_key: str, user_id: str, attributes: Optional[UserAttributes] = None

) -> Any:

""" Returns value for a variable attached to a feature flag.

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('get_feature_variable'))

return None

return self._get_feature_variable_for_type(project_config, feature_key, variable_key, None, user_id, attributes)

def get_feature_variable_boolean(

self, feature_key: str, variable_key: str, user_id: str, attributes: Optional[UserAttributes] = None

) -> Optional[bool]:

""" Returns value for a certain boolean variable attached to a feature flag.

variable_type = entities.Variable.Type.BOOLEAN

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('get_feature_variable_boolean'))

return None

return self._get_feature_variable_for_type( # type: ignore[no-any-return]

project_config, feature_key, variable_key, variable_type, user_id, attributes,

)

def get_feature_variable_double(

self, feature_key: str, variable_key: str, user_id: str, attributes: Optional[UserAttributes] = None

) -> Optional[float]:

""" Returns value for a certain double variable attached to a feature flag.

variable_type = entities.Variable.Type.DOUBLE

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('get_feature_variable_double'))

return None

return self._get_feature_variable_for_type( # type: ignore[no-any-return]

project_config, feature_key, variable_key, variable_type, user_id, attributes,

)

def get_feature_variable_integer(

self, feature_key: str, variable_key: str, user_id: str, attributes: Optional[UserAttributes] = None

) -> Optional[int]:

""" Returns value for a certain integer variable attached to a feature flag.

variable_type = entities.Variable.Type.INTEGER

project_config = self.config_manager.get_config()

if not project_config:

self.logger.error(enums.Errors.INVALID_PROJECT_CONFIG.format('get_feature_variable_integer'))

return None

return self._get_feature_variable_for_type( # type: ignore[no-any-return]

project_config, feature_key, variable_key, variable_type, user_id, attributes,

)

def get_feature_variable_string(

self, feature_key: str, variable_key: str, user_id: str, attributes: Optional[UserAttributes] = None

) -> Optional[str]:

""" Returns value for a certain string variable attached to a feature.