Skip to content

Instantly share code, notes, and snippets.

@jafayer
Last active July 8, 2023 19:19
Show Gist options
  • Save jafayer/c28f09b99055147ea1ee30fe34df2528 to your computer and use it in GitHub Desktop.
Save jafayer/c28f09b99055147ea1ee30fe34df2528 to your computer and use it in GitHub Desktop.
A demo of how decorators could be used for Parsons API Connector extended functionality
import time
import random
import requests
def poll(interval, falsy_condition=False):
def wrapped(func):
def inner(*args):
result = func(*args)
while result == falsy_condition:
print("Nope!", result)
time.sleep(interval)
result = func(*args)
return result
return inner
return wrapped
def exponential_backoff(base, falsy_condition=False, with_jitter=True, jitter_range=(0, 2)):
def wrapped(func):
def inner(*args):
count = 0
result = func(*args)
while result == falsy_condition:
num = base * 2 ** count
if with_jitter:
num = num + random.randint(*jitter_range)
print("NOPE", num)
time.sleep(num)
count += 1
result = func(*args)
return result
return inner
return wrapped
def datakey(key):
def wrapped(func):
def inner(*args):
result = func(*args)
return result.get(key)
return inner
return wrapped
@exponential_backoff(1)
def chance_exp(val, maxm):
"""
Allows you to completely abstract away backoff logic
provided your method returns a predicable response
in case of failure
"""
return random.randint(0, maxm) == val
@poll(1)
def chance_poll(val, maxm):
"""
Same with polling. Logic is left intact without having
to worry about retries provided failure is predictable
"""
return random.randint(0, maxm) == val
@poll(1, falsy_condition=None)
@datakey('products')
def get_products(url, val, max):
"""
It gets more complicated, but you can even stack them!
You DO need to peek a bit into the abstraction to understand
how to do this, though, which I don't love.
Note that in this implementation, `falsy_condition` is None in @poll.
This is because @datakey happens FIRST: @poll wraps around @datakey.
The function get_products returns a response which MUST be subscriptable (a dict),
datakey tries response.get(datakey), and returns None if it's not there
"""
randint = random.randint(0, max)
if randint == val:
response = requests.get(url)
return response.json()
else:
return {}
@datakey('products')
@poll(1, falsy_condition={})
def get_products_reverse_decorators(url, val, max):
"""
Note how in THIS identical implementation with the decorators reversed,
`falsy_condition` in @poll is {} instead of None.
That's because the @poll happens first, and only once the response exists
does it pass it to @datakey. This changing of arguments depending on the order
of decorators used is pretty leaky.
Another odd behavior is that with @poll last, your function can return anything
in the falsy case as long as `falsy_condition` matches it.
For example, this function could instead have `else: return False`
as long as `falsy_condition=False`in @poll.
This is NOT the case with the first implementation. With @poll above @datakey,
@datakey gets the response from get_products and tries to extract data, then
@poll gets the response from a failed response.get() in @datakey, which will always be None.
With @datakey closer to the method, you MUST return a dict for @datakey.
"""
randint = random.randint(0, max)
if randint == val:
response = requests.get(url)
return response.json()
else:
return {}
print("exponential backoff")
print(chance_exp(1,2))
print("polling")
print(chance_poll(1,5))
print("get_products with @poll then @datakey")
print(get_products('https://dummyjson.com/products', 2, 5))
print("get_products with @datakey then @poll")
print(get_products_reverse_decorators('https://dummyjson.com/products', 2, 5))
class Thing:
"""
An example class implementation that leverages a data_key passed in at creation
"""
def __init__(self, data_key):
self.data_key = data_key
def data_key(data_key_override=None):
def decorator(func):
def inner(self, *args):
result = func(self, *args)
return result.get(self.data_key if not data_key_override else data_key_override)
return inner
return decorator
@data_key()
def get(self, url):
print(self.data_key)
return requests.get(url).json()
@data_key('data')
def get_override(self, url):
print(self.data_key)
return requests.get(url).json()
t = Thing(data_key='products')
print(t.get('https://dummyjson.com/products'))
print(t.get_override('https://dummyjson.com/products'))
@jafayer
Copy link
Author

jafayer commented Jul 8, 2023

Just noting I updated the gist to add @data_key functionality and a more built-out demo at the end.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment