Last active
July 8, 2023 19:19
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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')) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Just noting I updated the gist to add @data_key functionality and a more built-out demo at the end.