Reference
Start SDK
To initialize and start the SDK, just call the start() function of the Flagship class in the most appropriate location for your application.
def start(env_id, api_key, configuration=None)
def start(env_id, api_key, configuration=None)from flagship import Flagship
Flagship.start('_YOUR_ENV_ID_', '_YOUR_API_KEY_')env_id
Yes
str
Environment id provided by Flagship.
api_key
Yes
str
Api authentication key provided by Flagship.
configuration
No
DecisionApi / Bucketing (FLagshipConfig)
Flagship configuration (DecisionApi or Bucketing)
Configure your Python SDK
These classes aim to help you to configure the SDK via the following two available config implementations: DecisionApi and Bucketing.
DecisionApi (default)
Run the SDK in DecisionApi mode. The campaign assignments and targeting validation take place server-side. In this mode, each call to the fetchFlags method to refresh the flags will create an HTTP request.
from flagship import Flagship
from flagship.config import DecisionApi
Flagship.start('_YOUR_ENV_ID_', '_YOUR_API_KEY_', DecisionApi(
timeout=3000,
log_level=LogLevel.ALL,
status_listener=CustomStatusListener(),
log_manager=CustomLogManager()
))timeout
int
2000
Specifies timeout for decision api requests in milliseconds
log_level
LogLevel
LogLevel.ALL
Specifies a log level to filter logs emitted by the SDK.
status_listener
StatusListener
None
Specifies a callback to be notified when the SDK status has changed.
log_manager
LogManager
FlagshipLogManager
Specifies a custom implementation of LogManager in order to receive logs from the SDK.
tracking_manager_config
TrackingManagerConfig
Default Tracking Manager Config
Specifies a custom tracking manager configuration. See TrackingManagerConfig section
cache_manager
CacheManager
None
Specifies a custom cache manager implementation. See CacheManager section
Bucketing
When the SDK is running in Bucketing mode, the SDK downloads all the campaigns configurations at once in a single bucketing file so that variation assignment can be computed client-side by the SDK. This bucketing file is stored in cache and will only be downloaded again when campaign configurations are modified in the Flagship interface.
from flagship import Flagship
from flagship.config import Bucketing
Flagship.start('_YOUR_ENV_ID_', '_YOUR_API_KEY_', Bucketing(
timeout=3000,
log_level=LogLevel.ALL,
status_listener=CustomStatusListener(),
polling_interval=2000,
log_manager=CustomLogManager()
))timeout
int
2000ms
Specifies timeout for decision api requests in milliseconds
log_level
LogLevel
LogLevel.ALL
Specifies a log level to filter logs emitted by the SDK.
status_listener
StatusListener
None
Specifies a callback to be notified when the SDK status has changed.
log_manager
LogManager
FlagshipLogManager
Specifies a custom implementation of LogManager in order to receive logs from the SDK.
polling_interval
int
60000ms
Defines the time interval between two bucketing updates in milliseconds.
tracking_manager_config
TrackingManagerConfig
Default Tracking Manager Config
Specifies a custom tracking manager configuration. See TrackingManagerConfig section
cache_manager
CacheManager
None
Specifies a custom cache manager implementation. See CacheManager section
Create new visitors
The visitor instance is an helper object that lets you manage the context and campaigns for a visitor identified by a unique ID.
The visitor context is a property dataset that defines the current visitor of your app. This dataset is sent and used by the Flagship Decision API or the SDK as targeting criteria for campaign assignments.
For example, if you want to enable or disable a specific feature based on VIP status, you would pass this attribute as a key-value pair in the visitor context so that the Decision API can enable or disable the corresponding feature flag for the user.
def new_visitor(visitor_id, **kwargs)
def new_visitor(visitor_id, **kwargs)visitor = Flagship.new_visitor('your_visitor_unique_id',
instance_type=Visitor.Instance.NEW_INSTANCE,
authenticated=True,
consent=True,
context={
'vip':True
}
)visitor_id
Required
String
/
Visitor unique identifier
Kwargs
instance_type
optional
Instance
NEW_INSTANCE
This class specifies how Flagship SDK should handle the newly created visitor instance. See Visitor.Instance
authenticated
optional
Bool
False
Bool that Specifies if the visitor is authenticated (True) or anonymous (False). See Keep experience continuity
consent
optional
Bool
True
Bool that Specifies if the visitor has consented for personal data usage. When false some features will be deactivated, cache will be deactivated and cleared.
context
optional
Dict
{}
Dict that specifies visitor initial context key / values used for targeting. Context keys must be String, and values types must be one of the following : Number, Boolean, String. See Managing visitor context
Visitor Instance
Visitor.Instance is an enum class that help you to define how the SDK should handle the visitor reference. There are two types:
SINGLE_INSTANCE
The newly created visitor instance will be returned and saved into the Flagship singleton. Call Flagship.getVisitor() to retrieve the instance. This option should be adopted on applications that handle only one visitor at the same time.
NEW_INSTANCE
The newly created visitor instance wont be saved and will simply be returned. Any previous visitor instance will have to be recreated. This option should be adopted on applications that handle multiple visitors at the same time.
Manage visitor consent
The Visitor class provides a method to let you manage visitor consent for data privacy usage. When False, campaign exposition and hits will be disabled and cache cleared until consent is passed True again.
def set_consent(self, consent)
def set_consent(self, consent)visitor.set_consent(True)consent
Required
Bool
Specify if the visitor has consented for personal data usage. When false some features will be deactivated, cache will be deactivated and cleared.
🚧 Consent hits
Consent hit requests will still be enabled in order to clear server-side cached data.
Target visitors by updating its context
The user context is a property dataset that defines the current user of your app. This dataset is sent and used by the Flagship Decision API or by the SDK in Bucketing mode as targeting criteria for campaign assignment.
The following method from the visitor instance allows you to set new context values matching the given keys.
from flagship import Flagship, Visitor
#It is possible to pass initial context at visitor creation.
visitor = Flagship.new_visitor('your_visitor_unique_id', context={'vip':True})
#Later it is possible to update visitor context when you have more information.
last_purchase_info = {
'last_purchase_date':'07-04-2023',
'last_purchase_amount':99.9
}
visitor.update_context(last_purchase_info)
#Once the context is updated, you need to fetch the flags again so campaign assignments
# are also updated.
visitor.fetch_flags()def update_context(self, context)
def update_context(self, context)context
Required
Tuple/Dict
Update the visitor context values, matching the given keys, used for targeting. A new context value associated with this key will be created if there is no previous matching value. Context keys must be Str, and values types must be one of the following : Number, Bool, Str.
Predefined visitor context keys
The Flagship SDK contains predefined user context keys so the Flagship plateform will be able to recognize them.
You can overwrite these keys at any time, only the ones starting by 'FLAGSHIP' are reserved and cant be overridden.
All possible predefined keys are contained in the FlagshipContext enum class and are listed below:
DEVICE_LOCALE
str
Define the current device locale in the visitor context. (must be a iso3 code str)
DEVICE_TYPE
DeviceType
Define the current device type in the visitor context.
DEVICE_MODEL
str
Define the current device model (Google Pixel 3) in the visitor context
LOCATION_CITY
str
Define the current city location in the visitor context.
LOCATION_REGION
str
Define the current region location in the visitor context.
LOCATION_COUNTRY
str
Define the current country location in the visitor context.
LOCATION_LAT
float
Define the current latitude location in the visitor context.
LOCATION_LONG
float
Define the current longitude location in the visitor context.
OS_NAME
str
Define the current OS name in the visitor context.
OS_VERSION
str
Define the current OS version name in the visitor context.
CARRIER_NAME
str
Define the current carrier name in the visitor context.
INTERNET_CONNECTION
str
Define the current connection type (ex Edge/3G/4G/5G/Wifi) in the visitor context.
APP_VERSION_NAME
str
Define the current app version in the visitor context.
APP_VERSION_CODE
int
Define the current app version in the visitor context.
INTERFACE_NAME
str
Define the current interface name or URL in the visitor context.
FLAGSHIP_CLIENT
str
Reserved by Flagship. Specifies the current client SDK stack.
FLAGSHIP_VERSION
str
Reserved by Flagship. Specifies the current version of the SDK.
Fetch flags
The fetch_flags() method of the visitor instance automatically updates the campaign assignments according to the current user context, retrieves applicable flags and store them in the visitor instance.
When the SDK is configured with the DecisionApi decision mode, flags will be fetched from the Flagship Decision API via an http request. When configured with Bucketing decision mode, flags will be fetched locally via de decision file.
def fetch_flags(self)
def fetch_flags(self)from flagship import Flagship, Visitor
visitor = Flagship.new_visitor('your_visitor_unique_id', context={'vip':true})
# Get the flags from assigned campaign that match the given visitor context.
visitor.fetch_flags()Use your flags
Once the campaign has been assigned and fetched, all the flags are stored in the SDK. You can retrieve these flags using the following get_flag function from the Visitor instance:
def get_flag(self, key, default)
def get_flag(self, key, default)from flagship import Flagship, Visitor
visitor = Flagship.new_visitor('your_visitor_unique_id', context={'vip':True})
visitor.fetch_flags()
#At this point, flags are available in the visito instance.
flag = visitor.get_flag("vip_feature", False)
#Get the value current flag value.
flag_value = flag.value()This function will return a flag object containing the current value returned by Flagship and the associated campaign metadata. If the key is not found an empty Flag object with the default value will be returned.
key
str
Flag key associated to the flag.
default
str | bool | Number | Dict | Array
Fallback value to us
Get Flags values
def value(self, user_exposed=True)
def value(self, user_exposed=True)Returns the current value for this flag or the default value when:
The flag doesn't exist.
The the current value and default value types are different.
The flag exists from the reference variation and no value have been configured on the platform.
flag = visitor.get_flag("vip_feature", False)
#Get the value current flag value.
flag_value = flag.value()
if flag_value:
#display the feature for vip users.key
str
Flag key associated to the flag.
default
str | bool | Number | Dict | Array
Fallback value to use
Expose Flags to user
Flag expositions are used to calculate visits on your campaign reporting.
By default when the value() method is called, the SDK considers that the visitor have seen the effets of this Flag and automatically sends a hit event to our data collection.
If you plan to use the flag value later, passing False will allow you to report flag exposition manually via the visitor_exposed method whenever your visitor really see the effect of the flag.
def visitor_exposed(self)
def visitor_exposed(self)flag = visitor.get_flag("vip_feature", False)
####
flag_value = flag.value() # Visitor exposition will automatically be reported.
if flag_value:
#display the feature for vip users.
####
flag_value = flag.value(False) # Visitor exposition won't automatically be reported.
# ...
if flag_value:
#display the feature for vip users.
flag.visitor_exposed() # Visitor exposition will only be reported at this point.
Check if a Flag exists
Call the exists()method to check if a Flag exists in the Flagship SDK. This will be useful when the default value is returned from the value() method to know it is due to the reference variation or because the Flag do not exists.
def exists(self)
def exists(self)This function will return:
False if the Flag do not exists in Flagship or the value type doesn't match the given default value type.
True otherwise.
Retrieve campaign metadata
You may need to send campaign IDs or variation IDs to a third party for reporting and/or analytics purposes. It is possible to retrieve campaigns metadata for a specific Flag.
def metadata(self)
def metadata(self)This method returns a FlagMetadata object.
flag = visitor.get_flag("vip_feature", False)
flag_metadata = flag.metadata()
campaign_id = flag_metadata.campaign_id
variation_group_id = flag_metadata.variation_group_id
variation_id = flag_metadata.variation_id
campaign_slug = flag_metadata.campaign_slug
campaign_type = flag_metdata.campaign_type
is_reference = flag_metadata.is_referenceclass FlagMetadata
class FlagMetadatacampaign_id
str
Flag campaign unique identifier.
variation_group_id
str
Flag variation group unique identifier.
variation_id
str
Flag variation unique identifier.
campaign_slud
str
Flag campaign slug.
campaign_type
str
Flag campaign type.
is_reference
bool
Is the Flag from variation reference.
Send tracking hits
This section helps you track your users and learn how to build hits in order to feed campaign goals. For more information about our measurement protocol, read our Universal Collect documentation.
There are 5 types of hits: Page, Screen, Transaction, Item and Event.
Every hits emitted by visitors are managed by a composant called TrackingManager which gather all the visitors hits in a pool, batch them and send them at regular time intervals or when the pool max size is reached.
The TrackingManager service aims to save bandwidth traffic usage with its batch processing, but also aims to prevent any data loss as it will try to re-send hits that have failed due to network issue or save them into cache when it is necessary.
There are five different types of Hits available:
Page
Screen
Transaction
Item
Event
Hits must all be built and sent with the send_hit function from the Visitor instance
Send Hits
from flagship.hits import Page
visitor.send_hit(Page("https://docs.developers.flagship.io/"))def send_hit(self, hit)
def send_hit(self, hit)Send hits as objectives in your campaign reporting.
hit
Hit
Hit to track
Page
This hit can be sent each time a visitor visits a web page, or a local page or in an embedded web view.
from flagship.hits import Page
visitor.send_hit(Page("https://docs.developers.flagship.io/"))class Page(Hit)
class Page(Hit)origin
String
Valid URL.
Screen
This hit can be sent each time a visitor arrives on an app interface.
from flagship.hits import Screen
visitor.send_hit(Screen("your_screen_name"))class Screen(Hit)
class Screen(Hit)origin
String
Interface name.
Event
This hit can be used for
any event (e.g. Add To Cart click, newsletter subscription).
class Event(Hit)
class Event(Hit)from flagship.hits import Event, EventCategory
visitor.send_hit(Event(EventCategory.ACTION_TRACKING, "cta_proceed_cart")
.with_event_label('Click basket payment')
.with_event_value(3))category
EventCategory
Specifies the category of your event. NOTE: This value must be either 'ACTION_TRACKING' or 'USER_ENGAGEMENT'.
action
String
Event name that will also serve as the KPI that you will have inside your reporting. Learn more
with_event_label()
label
str
Set the event description.
with_event_value()
value
int
Set a number value to your event.
Transaction
This hit can be sent when a user complete a transaction.
class Transaction(Hit)
class Transaction(Hit)from flagship.hits import Transaction
visitor.send_hit(Transaction("#309830", "purchases")
.with_currency("EUR")
.with_item_count(3
.with_payment_method("credit_card")
.with_shipping_cost(14.99)
.with_shipping_method("Express delivery")
.with_taxes(240.00)
.with_total_revenue(1200.00)
.with_coupon_code("#SAVE10"))transaction_id
str
Unique transaction ID.
affiliation
str
Affiliation kpi name to report.
with_currency()
currency
str
Set the local currency for all transaction currency values. Value should be a valid ISO 4217 currency code.
with_item_count()
item_nb
int
Set the number of items purchased.
with_payment_method()
payment
str
Indicate the payment method.
with_shipping_cost()
shipping
float
Specifies the total shipping cost of the transaction.
with_shipping_method()
shipping_method
str
Indicate the shipping method.
with_taxes()
taxes
float
Specifies the total tax of the transaction.
with_total_revenue()
revenue
float
Specifies the total revenue associated with the transaction. This value should include any shipping or tax costs.
with_coupon_code()
coupon
str
Set the coupon code associated with the transaction.
Item
This hit can be used to link an item with a transaction. It must be sent after the corresponding transaction hit.
class Item(Hit)
class Item(Hit)from flagship.hits import Item
visitor.send_hit(Item("#309830", "RTX4080", "#cg_rtx_40802023")
.with_item_category("hardware")
.with_item_quantity(1)
.with_price(1200.00))transaction_id
str
The unique transaction ID to link with this item
product_name
str
Name of the product.
product_sku
str
Product stock keeping unit.
with_price()
price
float
Specifies the price for a single item / unit.
with_item_quantity()
item_quantity
int
Specifies the number of items purchased.
with_item_category()
category
str
Specifies the category which the item belongs to.
Hits common builder functions
with_ip()
ip
str
The IP address of the user. This should be a valid IP address in IPv4 or IPv6 format. It will always be anonymized.
with_resolution(
witdh, height
int
Set the user's device resolution. \n :param width: width in pixels. Max length 10 Bytes. Min value 0. \n :param height: height in pixels. Max length 10 Bytes. Min value 0.
with_session_number()
number
int
Number of the current session for the current visitor.
with_locale()
locale
str
Set locale of the user's device. Max length 20 Bytes.
Tracking Manager configuration
TrackingManager gathers all the hits emitted by visitors in a pools before sending them into batches to our API to limit bandwidth usage. This composant is configurable at SDK start method, this will allow you to specify the frequency of sending batches and the max pool size so you can adapt it to your visitors and hit volumetry.
When a batch fail to be sent for example due to a network failure, all of the hits inside the failed batch will be added back into the pool for future iteration.
To prevent any data loss, configure a custom cache implementation, see CacheManager, in order for the TrackingManager to be able to cache the remaining Hits in the pool when your app close, and load them when it needs to try to send them again. The CachingStrategy will help you to specify at which frequency the cache will be updated.
from flagship import Flagship
from flagship.config import Bucketing
from flagship.tracking_manager import TrackingManagerConfig, CacheStrategy
Flagship.start('_YOUR_ENV_ID_', '_YOUR_API_KEY_', DecisionApi(
tracking_manager_config=TrackingManagerConfig,
max_pool_size=10,
time_interval=2000,
cache_strategy=CacheStrategy.PERIODIC_CACHING))max_pool_size
int
Default is 20
Define the minimum number of hits the pool must reach to automatically batch all hits in the pool and send it \n \nNote: \n- Having a very large max_pool_size can lead to performance issues
time_interval
int
10000 ms
Define a regular interval in milliseconds to trigger batch processing \n \nNote: \n- The process will batch all hits from the pool whether max_pool_size is reached or not.
cache_strategy
CacheStrategy
CONTINUOUS_CACHING | \nPERIODIC_CACHING
Define the strategy that will be used for hit caching. See \nCacheStrategy
Hits caching strategy
The CacheStrategy enum class specifies the hits caching strategy to adopt into the TrackingManager and relies on the HitCacheImplementation class that must be implemented in the CacheManager. Depending on the strategy the TrackingManager will request the CacheManager to CACHE, LOOK-UP or FLUSH hits from the linked database.
There are two available values for the CacheStrategy:
CONTINUOUS_CACHING
Hits will be continuously cached and flushed from the database. The database linked to the HitCacheImplementation class implemented in the provided CacheManager will be required each time time a hit is added or flushed from the TrackingManager pool.
PERIODIC_CACHING
Hits will be cached and flushed from the database periodically. The database linked to the HitCacheImplementation class implemented in the provided CacheManager will be required to cache/look-up/flush hits at regular time intervals. The time intervals relies on the TrackingManager 'time_interval' option.
Manage SDK Cache
At some points the Flagship SDK will require to cache and retrieve some information about visitors and hits.
Cached data will be used to:
Prevent re-allocation in bucketing mode if you have changed your traffic allocation in bucketing mode.
Handle Hits that failed to be sent due to network failure.
Be compatible with some features in Bucketing mode, like Dynamic allocation and Progressives Roll-out.
The Python SDK provides a way for you to customize where it will cache the needed data tanks to an interface to implement. You might want to pass your custom cache interface implementation to save and load data from an existing Redis or Memcached service.
Customize cache manager
Set your custom cache implementation in the start method. See configuration :
from flagship import Flagship
from flagship.config import DecisionApi
from flagship.cache_manager import CacheManager, VisitorCacheImplementation, HitCacheImplementation
class CustomCacheManager(CacheManager, VisitorCacheImplementation, HitCacheImplementation):
def __init__(self):
super().__init__()
def open_database(self, env_id):
#Create or open your database
pass
def close_database(self):
#close your database
pass
async def cache_visitor(self, visitor_id, data):
#Upsert visitor information corresponding to the given id into your database here.
pass
async def lookup_visitor(self, visitor_id):
#Load and return visitor information corresponding to the given id from your database here.
pass
async def flush_visitor(self, visitor_id):
#Delete visitor information corresponding to the given id from your database here.
pass
def cache_hits(self, hits):
#Cache the given hits into your database here.
pass
async def lookup_hits(self):
#Load and return all the cached hits from your database here.
pass
def flush_hits(self, hits_ids):
#Delete hits corresponding to the given id from your database here.
pass
def flush_all_hits(self):
#Delete all the hits cached hits from your database here.
pass
Flagship.start(env_id, api_key, DecisionApi(cache_manager=CustomCacheManager())There are two cache interfaces to implement:
Visitor cache interface
TheVisitorCacheImplementation class aims to provide an interface between the Flagship SDK and an existing database in order for the SDK to store visitors data. It defines the methods that will be called to handle the cache mechanism.
VisitorCacheImplementation
Cache visitor
This method is called when the Flagship SDK needs to save visitor's data into cache.
async def cache_visitor(self, visitor_id: str, data: dict)
visitor_id
str
Identifier of the visitor whose data must be cached.
data
dict
Visitor's data to be cached.
Lookup visitor
This method is called when the Flagship SDK needs to load visitor's data from cache.
async def lookup_visitor(self, visitor_id: str)
visitor_id
str
Identifier of the visitor whose cache must be loaded.
Flush visitor
This method is called when the Flagship SDK needs to flush visitor's data from cache.
async def flush_visitor(self, visitor_id: str)
visitor_id
str
Identifier of the visitor whose cache must be flushed.
Hit cache interface
The HitCacheImplementation class aims to provide an interface between the Flagship SDK and an existing database in order to store hits data in case of error or network failure. It defines the methods that will be called by the SDK to handle the cache mechanism depending on the strategy.
HitCacheImplementation
Cache hits
This method is called when the Flagship SDK needs to save visitors hits into cache.
def cache_hits(self, hits)
hits
dict
Dictionary of hits that need to be saved into cache.
Lookup hits
This method is called when the Flagship SDK needs to load visitors hits from the cache.
async def lookup_hits(self)
Flush hits
def flush_hits(self, hits_ids)
hits_ids
dict
Hits ids that need to be flushed from cache.
Flush all hits
def flush_all_hits(self)
This method is called when the Flagship SDK needs to flush all the hits from cache.
Keep visitors experience continuity
Dealing with anonymous and logged-in users, experience continuity allows you to maintain consistency between sessions and devices. See Experience continuity concept for more details.
🚧 Enable experience continuity feature
The use of this feature requires having the option enabled in your Flagship account beforehand. Select the desired environment and go to /settings/environment-settings and enable the "Experience Continuity" option.
Authenticate a visitor
There are 2 ways to authenticate a visitor and let the SDK knows that your visitor is authenticated:
Set the
authenticatedto True when creating a new visitor. See create new visitor.Use
authenticate()function ofVisitorinstance.
def authenticate(self, visitor_id)
This function will indicate to the SDK that the anonymous visitor is now authenticated, this will insure to keep the same experience.
visitor.authenticate("visitor_unique_id")visitor_id
str
Current authenticated visitor unique identifier.
🚧 Update visitor flags
Because the visitor have been now identified, it is required to call the fetch_flags()function of the visitor instance so his flags will be updated accordingly. See fetching flags.
Un-Authenticate a visitor
def unauthenticate(self)
visitor.unauthenticate()This function will indicate to the SDK the authenticated visitor is now un-authenticated and back to anonymous.
🚧 Update visitor flags
Because the visitor have been now identified, it is required to call the fetch_flags()function of the visitor instance so his flags will be updated accordingly. See fetching flags.
Customize the logs of the SDK
The SDK provides a way for you to intercept and/or customize the logs emitted. It is possible to pass your own implementation of the LogManager class in the configuration.
The log_level option of the start method will filter the emitted logs.
from flagship import Flagship, LogLevel
from flagship.config import DecisionApi
from flagship.decision_mode import DecisionMode
from flagship.log_manager import LogManager, FlagshipLogManager
class CustomLogManager(LogManager):
def log(self, tag, level, message):
pass
def exception(self, exception, traceback):
pass
Flagship.start("_env_id_", "_api_key_", DecisionApi(
timeout=3000,
log_level=LogLevel.DEBUG,
status_listener=CustomStatusListener(),
polling_interval=2000,
log_manager=CustomLogManager()
))Get the SDK status
The SDK can have different status states.
Last updated
Was this helpful?