Python SDK

pip install amplitude-analytics

from amplitude import Amplitude
client = Amplitude(AMPLITUDE_API_KEY)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

client.flush()

client.add(plugin_obj)

client.remove(plugin_obj)

client.shutdown()

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

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