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)

from flagship import Flagship

Flagship.start('_YOUR_ENV_ID_', '_YOUR_API_KEY_')

📘

Info

You can find your env_id and your api_key on your Flagship account, in Parameters > Environment & Security.

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

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

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

visitor = Flagship.new_visitor('your_visitor_unique_id',
                               instance_type=Visitor.Instance.NEW_INSTANCE,
                               authenticated=True,
                               consent=True,
                               context={
                                 'vip':True
                               }
                              )

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:

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)

visitor.set_consent(True)

🚧

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)

🚧

Fetch flags

It is necessary to call fetch_flags() method after having the new context updated so campaigns will also be updated accordingly.

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:

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)

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)

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.

Get Flags values

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.

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)

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)

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)

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_reference

class FlagMetadata

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:

from flagship.hits import Page

visitor.send_hit(Page("https://docs.developers.flagship.io/"))

def send_hit(self, hit)

Send hits as objectives in your campaign reporting.

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)

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)

Event

This hit can be used for

any event (e.g. Add To Cart click, newsletter subscription).

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

Transaction

This hit can be sent when a user complete a transaction.

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

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)

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

Hits common builder functions

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

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:

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.

📘

Sqlite implementation

The SDK already embeds a predefined Sqlite implementation to cache data locally but it is will only be suitable for services that handle one visitor at a time like python apps, scripts, IOT.

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)

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)

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)

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)

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)

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 and SDK implementation 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:

1. Set the authenticated to True when creating a new visitor. See create new visitor.
2. Use authenticate() function of Visitor instance.

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

🚧

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.