from __future__ import absolute_import
from __future__ import print_function
import time
import swagger_client
import inspect
import tenacity
import klaviyo_sdk.custom_retry as custom_retry
import json
import base64
import requests
class Client:
def __init__(self, api_key,test_host=None, max_delay=60, max_retries=3):
configuration = swagger_client.Configuration()
self.api_key = api_key
self.retry_codes = [429,503,504,524]
self.retry_logic = tenacity.retry(
reraise=True,
retry=custom_retry.retry_if_qualifies(self.retry_codes),
wait=tenacity.wait.wait_random_exponential(multiplier = 1, max = max_delay/max_retries),
stop=tenacity.stop.stop_after_attempt(max_retries)
)
configuration.api_key["api_key"] = api_key
if test_host:
configuration.host = test_host
subclient_names = [item_name for item_name in dir(swagger_client.api) if inspect.isclass(getattr(swagger_client.api,item_name)) and 'with_http_info' not in item_name]
for subclient_name in subclient_names:
setattr(self,subclient_name,eval(f'swagger_client.{subclient_name}(swagger_client.ApiClient(configuration))'))
subclient = eval(f'self.{subclient_name}')
for attribute_name in dir(subclient):
if f'{attribute_name}_with_http_info' in dir(subclient):
endpoint = eval(f'subclient.{attribute_name}')
endpoint = self.retry_logic(endpoint)
setattr(subclient,attribute_name,endpoint)
self.TrackIdentifyApi.track_post = self.post_update(self.TrackIdentifyApi.track_post)
self.TrackIdentifyApi.identify_post = self.post_update(self.TrackIdentifyApi.identify_post)
self.TrackIdentifyApi.track_get = self.get_update(self.TrackIdentifyApi.track_get)
self.TrackIdentifyApi.identify_get = self.get_update(self.TrackIdentifyApi.identify_get)
self.ProfilesApi.update_profile = self.update_profile_fix(self.ProfilesApi.update_profile)
# last step: drop 'Api' suffix from sublient name
for subclient_name in subclient_names:
setattr(self,subclient_name[:-3],eval(f'self.{subclient_name}'))
def is_error(self, status):
return not (200 <= status <= 299)
def update_profile_fix(self, func):
def wrapped_func(person_id='PERSON_ID', params={}):
url = f"https://a.klaviyo.com/api/v1/person/{person_id}"
querystring = {"api_key":self.api_key}
for key in params:
querystring[str(key)] = str(params[key])
headers = {
"Accept": "application/json",
"user-agent" : "klaviyo-python-sdk/1.0.4.20220329"
}
response = requests.request("PUT", url, headers=headers, params=querystring)
if self.is_error(response.status_code):
e = swagger_client.rest.ApiException(status=response.status_code, reason=response.reason, http_resp=response)
raise(e)
return response.json()
return wrapped_func
def post_update(self, func):
def wrapped_func(data={}):
if type(data) is not str:
data = json.dumps(data)
return func(data=data)
return wrapped_func
def get_update(self, func):
def wrapped_func(data={}):
if type(data) is dict:
json_string = json.dumps(data)
utf = json_string.encode('utf-8')
data = base64.b64encode(utf)
return func(data)
elif type(data) is str:
utf = data.encode('utf-8')
data = base64.b64encode(utf)
return func(data)
elif type(data) is bytes:
if b'{' in data:
data = base64.b64encode(data)
return func(data)
return wrapped_func
Public https://github.com/klaviyo/klaviyo-python-sdk/blob/main/klaviyo_sdk/wrapper.py
Type hints help humans and linters (like mypy) to understand what to expect "in" and "out" for a function. Not only it serves as a documentation for others (and you after some time, when the code is wiped from your "brain cache"), but also allows using automated tools to find type errors.
Dataclasses let you get rid of many boilerplate code, most often the "init hell": def __init__(self, a): self.a = a. With dataclasses, it's all done automatically!
Key constants of a class should be so-called "class variables" instead of using/defining them inside methods. If you create a child class and want to change those variables, you'll have to rewrite entire methods. With class vars, you only have to overwrite those class vars in child. Example: class MyClass: SOME_VALUE = 1 and class MySubclass(Class): SOME_VALUE = 2
RETRY_CODES = {429, 503, 504, 524}
Searching in list is slow - O(n). Searching in set is fast - O(1). Also, lists are mutable while sets are not, so if you want to indicate and ensure that some collection never changes - use sets. So [1, 2, 3] -> {1, 2, 3}.
You. Should. Never. Use. Eval.
Executing code from a variable is very dangerous. Almost always there is a better option than using eval. Usually all you need is either getattr() or google.
Setattr and getattr are helpful when you calculate some attribute names dynamically. However, you later, your IDE, and other devs may have troubles understanding which attributes you accessed. Always try to avoid using setattr and getattr. Dictionaries are usually a good replacement.
Some methods within a class don't depend on self variable. This is an indication that such methods should be classmethods. Use the @classmethod decorator.
If some value is hidden somewhere in the middle of code, it will be very hard to find it later. Better solutions: 1) use environment variables 2) use constants in module's top section 3) use class variables.
When writing decorators, it is a good practice to decorate them with @wraps(decorated_function) which does some nice things you usually don't think about when decorating: preserving name and docstring of the decorated function.
dict type has built-in methods: keys() - to get keys iterator; values() - to get values iterator; items() - to get (key, value) iterator. Use it.
Surprise! Python's requests module does not have a timeout by default! So if you don't use the timeout parameter when making a request and if remote resource hangs forever (or is very slow), your code will hang, too. Always use reasonable timeouts. A good value may be 5 seconds.
This code is not really needed or may be simplified
response = requests.put(...)
Each call of requests.get() (and other methods) creates a new connection to web resource. It is okay if you make a single request, but is very non-optimal when making subsequent requests. You should always use sessions which will hold connection between requests: session = requests.Session(); response = session.get(url)
requests module has built-in method to raise exception on bad response: response.raise_for_status(). If you need to perform some additional actions on bad response, use if not response.ok: ...
Idiomatic python type check is if isinstance(type, variable) or if isinstance((type1, type2), variable). Using type(variable) == SomeClass smells because it won't work if variable is a subclass of SomeClass.
Create new review request