Python SDK

The Python SDK lets you send events to Amplitude.

Install the SDK

Install amplitude-analytics with pip:

pip install amplitude-analytics

Initialize the SDK

Initialize the SDK before you instrument any events. The Amplitude project API key is required. You can also pass a config object in this call. Reuse the SDK client instance across requests after initialization.

from amplitude import Amplitude

client = Amplitude(AMPLITUDE_API_KEY)

Configure the SDK

def callback_func(event, code, message=None):
  # callback function that takes three input parameters
  # event: the event that triggered this callback
  # code: status code of request response
  # message: a optional string message for more detailed information

client.configuration.api_key = "new api key"
client.configuration.flush_max_retries = 5
client.configuration.logger = logging.getLogger(__name__)
client.configuration.min_id_length = 7
client.configuration.callback = callback_func
client.configuration.server_zone = "EU"
client.configuration.use_batch = True
client.configuration.server_url = "proxy url that forwarding the requests"
client.configuration.opt_out = False

Configure batching behavior

The SDK queues events from the track method in memory and flushes them in batches in the background. Customize batch behavior with flush_queue_size and flush_interval_millis. By default, the SDK runs in regular mode with serverUrl set to https://api2.amplitude.com/2/httpapi. To send large batches of data at a time, set use_batch to true. The SDK then sets serverUrl to the batch event upload API at https://api2.amplitude.com/batch. Regular mode and batch mode use the same flush queue size and flush intervals.

from amplitude import Amplitude

client = Amplitude(AMPLITUDE_API_KEY)

# Events queued in memory flush when the number of events exceeds the upload threshold
# Default value is 200
client.configuration.flush_queue_size = 100
# Events queue flushes every set number of milliseconds
# Default value is 10 milliseconds
client.configuration.flush_interval_millis = 20000 # 20 seconds

Track an event

Events represent how users interact with your application. For example, "Button Clicked" is a common event.

from amplitude import Amplitude, BaseEvent

client = Amplitude(AMPLITUDE_API_KEY)

# Track a basic event
# One of user_id and device_id is required
event = BaseEvent(event_type="Button Clicked", user_id="User Id")
client.track(event)

# Track events with optional properties
client.track(
    BaseEvent(
        event_type="type of event",
        user_id="USER_ID",
        device_id="DEVICE_ID",
        event_properties={
            "source": "notification"
        }
))

User properties

User properties describe your users, such as device details, preferences, or language. Set them to understand who performs an action in your app.

The Identify call sets a user's properties without sending an event. The SDK supports the operations set, set_once, unset, add, append, prepend, pre_insert, post_insert, remove, and clear_all on individual user properties. Declare the operations through the Identify interface. Chain multiple operations together in a single Identify object, then pass it to the Amplitude client to send to the server.

Set a user property

To set a user property: instantiate an Identify object, call Identify methods on it, then pass the object to the client.

from amplitude import Identify, EventOptions


identify_obj=Identify()
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Identify.set

Sets the value of a user property. For example, set a user's role.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.set("location", "LAX")
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Identify.set_once

Increments a user property by a numerical value. If the user property doesn't have a value yet, the SDK initializes it to 0 before incrementing. For example, track a user's travel count.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.add("travel-count", 1)
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Arrays in user properties

Use arrays as user properties. Set arrays directly, or use prepend, append, pre_insert, and post_insert to build an array.

Identify.prepend

Prepends one or more values to a user property array. If the user property doesn't have a value yet, the SDK initializes it to an empty list before prepending.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.prepend("visited-locations", "LAX")
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Identify.append

Appends one or more values to a user property array. If the user property doesn't have a value yet, the SDK initializes it to an empty list before appending.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.append("visited-locations", "SFO")
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Identify.pre_insert

Inserts one or more values at the beginning of a user property array, only if those values don't exist in the array yet. If the user property doesn't have a value yet, the SDK initializes it to an empty list before inserting. If the user property already has the value, the operation is a no-op.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.pre_insert("unique-locations", "LAX")
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Identify.post_insert

Inserts one or more values at the end of a user property array, only if those values don't exist in the array yet. If the user property doesn't have a value yet, the SDK initializes it to an empty list before inserting. If the user property already has the value, the operation is a no-op.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.post_insert("unique-locations", "SFO")
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Identify.remove

Removes one or more values from a user property array if they exist. If the value doesn't exist in the user property, the operation is a no-op.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.remove("unique-locations", "JFK")
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

Identify.clear_all

Removes all user properties from a user. Use clear_all with care because the operation is irreversible.

from amplitude import Identify, EventOptions

identify_obj=Identify()
identify_obj.clear_all()
client.identify(identify_obj, EventOptions(user_id="USER_ID"))

User groups

Amplitude supports assigning users to groups and running queries such as Count by Distinct on those groups. For example, group users by organization with an orgId: assign Joe to orgId 10 and Sue to orgId 15. In the Event Segmentation chart, select "..performed by" orgId to query the number of organizations that performed a specific event. Amplitude includes a group in the count when at least one member performed the event.

When you set groups, define a group_type and group_name. In the previous example, orgId is the group_type, and 10 and 15 are each a group_name. Another example uses sport as the group_type with group_name values like tennis and baseball. Use set_group() to designate which groups a user belongs to. This also sets group_type:group_name as a user property. The call overwrites any existing group_name value for that user's group_type, along with the corresponding user property value. group_type is a string. group_name can be a string or an array of strings to indicate that a user belongs to multiple groups. For example, if Joe is in orgId 10 and 16, the group_name is [10, 16].

# set group with single group name
client.set_group(group_type="org_id", group_name="15",
                 event_options=EventOptions(user_id="USER_ID"))

# set group with multiple group names
client.set_group(group_type="org_id", group_name=["15", "21"],
                 event_options=EventOptions(user_id="USER_ID"))

Set event-level groups with the groups attribute of events.

# set groups when initial a event instance
event = BaseEvent("event_type", "user_id", groups={"org_id": ["15", "21"]})

# set groups for an existing instance
event["groups"] = {"sport": "soccer"}

client.track(event)

Group properties

Use the Group Identify API to set or update the properties of specific groups. These updates only affect events going forward.

The group_identify() method accepts a group type string, a group name string, and an Identify object to apply to the group.

identify_obj=Identify()
identify_obj.set("locale", "en-us")
client.group_identify(group_type="org-id", group_name="15", identify_obj=identify_obj)

Track revenue

To track revenue for a user, use revenue() with the Revenue interface. Revenue instances store each transaction and let you define special revenue properties such as revenue_type and product_id. Amplitude's Event Segmentation and Revenue LTV charts use these properties. Pass Revenue instance objects into revenue to send them as revenue events to Amplitude. Amplitude then displays revenue data automatically. Use this approach to track both in-app and non-in-app purchases.

To track revenue from a user, call revenue each time a user generates revenue. For example, a user purchases 3 units of a product at 3.99 each.

from amplitude import Revenue

revenue_obj = Revenue(price=3.99,
                      quantity=3,
                      product_id="com.company.productId")
client.revenue(revenue_obj, EventOptions(user_id="USER_ID"))

The revenue interface

Flush

The flush method triggers the client to send buffered events.

client.flush()

Add

The add method adds a plugin to the Amplitude client instance. Plugins help process and send events. Learn more about plugins.

client.add(plugin_obj)

Remove

The remove method removes the given plugin from the client instance if it exists.

client.remove(plugin_obj)

Shutdown

Use the shutdown method to close the instance. A closed instance doesn't accept new events and tries to flush events left in the buffer. After flushing, the client instance shuts down running threads.

In version v1.1.1 and higher, the SDK automatically registers the shutdown method to run when the main thread exits.

client.shutdown()

Plugins

Plugins extend the Amplitude SDK's behavior. For example, modify event properties (enrichment type) or send data to third-party APIs (destination type). A plugin is an object with the methods setup() and execute().

Plugin.setup

Prepares the plugin for use. Takes a client instance as a parameter and returns None. Typical use: copy configuration from client.configuration or instantiate plugin dependencies. The SDK calls this method when you register the plugin through client.add().

Plugin.execute

Processes events. Takes an event instance as a parameter. For an enrichment plugin, returns the modified or enriched event. For a destination plugin, returns a map with the keys event (BaseEvent), code (number), and message (string). The SDK calls this method for each event instrumented through the client interface, including Identify, GroupIdentify, and Revenue events.

Plugin examples

Enrichment type plugin

This example plugin modifies each instrumented event by adding an increment integer to the event's event_id property.

from threading import Lock
from amplitude import Amplitude, EventPlugin, PluginType

class AddEventIdPlugin(EventPlugin):

    def __init__(self, start=0):
        super().__init__(PluginType.ENRICHMENT)
        self.current_id = start
        self.configuration = None
        self.lock = Lock()

    def setup(self, client):
        self.configuration = client.configuration

    def execute(self, event):
        with self.lock:
            event.event_id = self.current_id
            self.current_id += 1
        return event

client = Amplitude(AMPLITUDE_API_KEY)
client.add(AddInsertIdPlugin())

Destination type plugin

from amplitude import Amplitude, EventPlugin, DestinationPlugin, PluginType
import requests

class MyDestinationPlugin(DestinationPlugin):

    def __init__(self):
        super().__init__()
        # other init operations
        self.url = "api endpoint url"
        self.configuration = None

    def setup(self, client):
        # setup plugin using client instance
        # triggered by client.add() method
        super().setup(client)
        self.configuration = client.configuration

    def execute(self, event):
        # process event using plugins in this destination plugin instance
        event = self.timeline.process(event)
        # send event to customized destination
        payload = '{"key":"secret", "event": ' + str(event) + '}'
        requests.post(self.url, data=payload)
        self.configuration.logger.info("Event sent")


client = Amplitude(AMPLITUDE_API_KEY)
client.add(MyDestinationPlugin())