How to use customerio - 10 common examples

def setUp(self):
        self.cio = CustomerIO(
            site_id='siteid',
            api_key='apikey',
            host=self.server.server_address[0],
            port=self.server.server_port,
            retries=5,
            backoff_factor=0)

        # do not verify the ssl certificate as it is self signed
        # should only be done for tests
        self.cio.http.verify = False

def test_base_url(self):
        test_cases = [
            # host, port, prefix, result
            (None, None, None, 'https://track.customer.io/api/v1'),
            (None, None, 'v2', 'https://track.customer.io/v2'),
            (None, None, '/v2/', 'https://track.customer.io/v2'),
            ('sub.domain.com', 1337, '/v2/', 'https://sub.domain.com:1337/v2'),
            ('/sub.domain.com/', 1337, '/v2/', 'https://sub.domain.com:1337/v2'),
            ('http://sub.domain.com/', 1337, '/v2/', 'https://sub.domain.com:1337/v2'),
        ]

        for host, port, prefix, result in test_cases:
            cio = CustomerIO(host=host, port=port, url_prefix=prefix)
            self.assertEqual(cio.base_url, result)

'method': 'POST',
            'authorization': _basic_auth_str('siteid', 'apikey'),
            'content_type': 'application/json',
            'url_suffix': '/segments/1/remove_customers',
            'body': {'ids': ['1','2','3']},
        }))

        self.cio.remove_from_segment(segment_id=1, customer_ids=[1,2,3])

        with self.assertRaises(CustomerIOException):
            self.cio.remove_from_segment(None, None)

        with self.assertRaises(CustomerIOException):
            self.cio.add_to_segment(segment_id=1, customer_ids=False)

        with self.assertRaises(CustomerIOException):
            self.cio.add_to_segment(segment_id=1, customer_ids=[False,True,False])

def test_client_connection_handling(self):
        retries = self.cio.retries
        # should not raise exception as i should be less than retries and 
        # therefore the last request should return a valid response
        for i in range(retries):
            self.cio.identify(i, fail_count=i)

        # should raise expection as we get invalid responses for all retries
        with self.assertRaises(CustomerIOException):
            self.cio.identify(retries, fail_count=retries)

def add_device(self, customer_id, device_id, platform, **data):
        '''Add a device to a customer profile'''
        if not customer_id:
            raise CustomerIOException("customer_id cannot be blank in add_device")
        
        if not device_id:
            raise CustomerIOException("device_id cannot be blank in add_device")
        
        if not platform:
            raise CustomerIOException("platform cannot be blank in add_device")

        data.update({
            'id': device_id,
            'platform': platform,
        })
        payload = {'device': data }
        url = self.get_device_query_string(customer_id)
        self.send_request('PUT', url, payload)

def add_to_segment(self, segment_id, customer_ids):
        '''Add customers to a manual segment, customer_ids should be a list of strings'''
        if not segment_id:
            raise CustomerIOException("segment_id cannot be blank in add_to_segment")

        if not customer_ids:
            raise CustomerIOException("customer_ids cannot be blank in add_to_segment")
        
        if not isinstance(customer_ids, list):
            raise CustomerIOException("customer_ids must be a list in add_to_segment")

        url = '{base}/segments/{id}/add_customers'.format(base=self.base_url, id=segment_id)
        payload = {'ids': self._stringify_list(customer_ids)}
        self.send_request('POST', url, payload)

def add_device(self, customer_id, device_id, platform, **data):
        '''Add a device to a customer profile'''
        if not customer_id:
            raise CustomerIOException("customer_id cannot be blank in add_device")
        
        if not device_id:
            raise CustomerIOException("device_id cannot be blank in add_device")
        
        if not platform:
            raise CustomerIOException("platform cannot be blank in add_device")

        data.update({
            'id': device_id,
            'platform': platform,
        })
        payload = {'device': data }
        url = self.get_device_query_string(customer_id)
        self.send_request('PUT', url, payload)

def _stringify_list(self, customer_ids):
        customer_string_ids = []
        for v in customer_ids:
            if isinstance(v, str):
                customer_string_ids.append(v)
            elif isinstance(v, int):
                customer_string_ids.append(str(v))
            else:
                raise CustomerIOException('customer_ids cannot be {type}'.format(type=type(v)))
        return customer_string_ids

def add_device(self, customer_id, device_id, platform, **data):
        '''Add a device to a customer profile'''
        if not customer_id:
            raise CustomerIOException("customer_id cannot be blank in add_device")
        
        if not device_id:
            raise CustomerIOException("device_id cannot be blank in add_device")
        
        if not platform:
            raise CustomerIOException("platform cannot be blank in add_device")

        data.update({
            'id': device_id,
            'platform': platform,
        })
        payload = {'device': data }
        url = self.get_device_query_string(customer_id)
        self.send_request('PUT', url, payload)

def send_request(self, method, url, data):
        '''Dispatches the request and returns a response'''

        try:
            response = self.http.request(method, url=url, json=self._sanitize(data), timeout=self.timeout)
        except Exception as e:
            # Raise exception alerting user that the system might be
            # experiencing an outage and refer them to system status page.
            message = '''Failed to receive valid reponse after {count} retries.
Check system status at http://status.customer.io.
Last caught exception -- {klass}: {message}
            '''.format(klass=type(e), message=e, count=self.retries)
            raise CustomerIOException(message)

        result_status = response.status_code
        if result_status != 200:
            raise CustomerIOException('%s: %s %s' % (result_status, url, data))
        return response.text