reusables

Reusables – Part 2: Wrappers

Spice up your code with wrappers! In Python, a wrapper, also known as a decorator, is simply encapsulating a function within other functions.

@wrapper
def my_func(a, b=2):
    print(b)

@meta_decorator(foo="bar")
def my_other_func(**kwargs):
    print(kwargs)

In Reusables, all the wrappers take arguments, aka meta decorators, so you will at minimum have to end them with parens ()s. Let’s take a look at the one I use the most first.

time_it

import reusables

@reusables.time_it()
def test_func():
    import time, random
    time.sleep(random.randint(2, 5))

test_func()
# Function 'test_func' took a total of 5.000145769345911 seconds

time_it documentation
The message by default is printed. However it can also be sent to a log with a customized message. If log=True it will log to the Reusables logger, however you can also specify either a logger by name (string) or pass in a logger object directly.

reusables.add_stream_handler('reusables')

@reusables.time_it(log=True, message="{seconds:.2f} seconds")
def test_time(length):
    time.sleep(length)
    return "slept {0}".format(length)

result = test_time(5)
# 2016-11-09 16:59:39,935 - reusables.wrappers  INFO      5.01 seconds

print(result)
# slept 5

It’s also possible to capture time taken in a list.

@reusables.time_it(log='no_log', append=my_times)
def test_func():
    import time, random
    length = random.random()
    time.sleep(length)
    
for _ in range(10):
    test_func()
    
my_times
# [0.4791555858872698, 0.8963652232890809, 0.05607090172793505, 
# 0.03099917658380491,0.6567622821214627, 0.4333975642063024, 
# 0.21456404424395714, 0.5723555061358638, 0.0734819056269771, 
# 0.13208268856499217]

unique

The oldest wrapper in the Reusables library forces the output of a function to be unique, or else it will raise an exception or if specified, return an alternative output.

import reusables
import random


@reusables.unique(max_retries=100)
def poor_uuid():
    return random.randint(0, 10)


print([poor_uuid() for _ in range(10)])
# [8, 9, 6, 3, 0, 7, 2, 5, 4, 10]

print([poor_uuid() for _ in range(1000)])
# Exception: No result was unique

unique documentation

lock_it

A very simple wrapper to use while threading. Makes sure a function is only being run once at a time.

import reusables
import time


def func_one(_):
    time.sleep(5)


@reusables.lock_it()
def func_two(_):
    time.sleep(5)


@reusables.time_it(message="test_1 took {0:.2f} seconds")
def test_1():
    reusables.run_in_pool(func_one, (1, 2, 3), threaded=True)


@reusables.time_it(message="test_2 took {0:.2f} seconds")
def test_2():
    reusables.run_in_pool(func_two, (1, 2, 3), threaded=True)


test_1()
test_2()

# test_1 took 5.04 seconds
# test_2 took 15.07 seconds

log_exception

It’s good practice to catch and explicitly log exceptions, but sometimes it’s just easier to let it fail naturally at any point and log it for later refinement or debugging.

@reusables.log_exception()
def test():
    raise Exception("Bad")

# 2016-12-26 12:38:01,381 - reusables   ERROR  Exception in test - Bad
# Traceback (most recent call last):
#     File "<input>", line 1, in <module>
#     File "reusables\wrappers.py", line 200, in wrapper
#     raise err
# Exception: Bad

queue_it

Add the result of the function to the specified queue instead of returning it normally.

import reusables
import queue

my_queue = queue.Queue()


@reusables.queue_it(my_queue)
def func(a):
    return a


func(10)

print(my_queue.get())
# 10

New in 0.9

catch_it

Catch specified exceptions and return an alternative output or send it to an exception handler.

def handle_error(exception, func, *args, **kwargs):
    print(f"{func.__name__} raised {exception} when called with {args}")

@reusables.catch_it(handler=handle_error)
def will_raise(message="Hello"):
    raise Exception(message)


will_raise("Oh no!")
# will_raise raised Oh no! when called with ('Oh no!',)

retry_it

Once is never enough, keep trying until it works!

@reusables.retry_it(exceptions=(Exception, ), tries=10, wait=1)
def may_fail(dont_do=[]):
    dont_do.append("x")
    if len(dont_do) < 6:
        raise OSError("So silly")
    print("Much success!")

may_fail()
# Much success!

 

This post is a follow-up of Reusables – Part 1: Overview and File Management.

Exploit the work of others for profit! (Vega 64 Edition)

As I sit here, anxious for the new AMD Vega 64 to be released, I decide to keep myself busy writing some Python code….designed to text me as soon as a new “rx vega 64” search term showed up on Amazon (I have the patience of a child on Christmas Eve, so sue me.)

When writing code, I try to be as lazy efficient as possible. That means I look for other’s to do the hard part for me. Other people might phrase it more kindly like “don’t reinvent the wheel,” but let’s be real, you are receiving benefit for no cost. So next time a project saves your bacon, consider sending a little cash or cryptocoin to the dev(s). Or throw your hat into the open source community and provide dev work yourself, it’s a great way to learn a lot more about the coding community and gain a lot of experience along the way while still giving back.

Back to the Vega 64 stock tracking tool. It would totally be possible to do that all with the Python internals; using urllib and re to download and find stuff on the page, then using email to send a message to my phone’s SMS; but that would take forever, and is honestly stupid to do. There are much better tools for that at this point, like requests and BeautifulSoup , then using some gmail or other common email provider library.

But as Amazon is a rather big website with an API available, there are Python libraries for that API. There are also different ways to easily send a text message to yourself via online services, like twilo. In the end, I created the script using Python 3.6 on Windows (should be cross-platform compatible), and the libraries I used for this were:

python-amazon-simple-product-api
twilio
reusables
python-box

If you are interested in using this as well (comes as-is, no promises it works or won’t bite you) before using the script you will need to get AWS access keys and sign up for twilio, then fill in the appropriate variables at the top of the script.

from datetime import datetime
from time import sleep

from amazon.api import AmazonAPI
from twilio.rest import Client
from box import Box, BoxList
from reusables import setup_logger

amazon_access_key = "XXXXXXXXXXXXXXXXXXXX"
amazon_secret = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
amazon_associate_tag = "codecalamity-20" 

twilio_from = "+"  # Twilio phone number
twilio_to = "+"  # Phone number to text
twilio_key = "ACXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
twilio_secret = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"

search_product_keywords = "rx vega 64"
search_product_name_includes = "vega"
search_product_brand = "amd"
search_index = 'Electronics'
search_region = 'US'

log = setup_logger("vega")


def update(search, saved_data):
    new_prods = BoxList()
    for x in (product for product in search if
              search_product_name_includes in product.title.lower()
              and (search_product_brand in product.brand.lower() or
                   search_product_brand in product.manufacturer.lower())):
        if x.title not in saved_data:
            data = Box({"price": float(x.price_and_currency[0]),
                        "url": x.detail_page_url,
                        "updated": datetime.now().isoformat()})
            log.info(f"New item: {x.title} - {data}")
            new_prods.append((x.title, data))
            saved_data[x.title] = data

    saved_data.to_json(filename="products.json")
    return new_prods


def format_message(new_prods):
    product_list = [f"{prod[0]}-{prod[1].price}-{prod[1].url}"
                    for prod in new_prods]
    return f"New Prods: {', '.join(product_list)}"


def send_message(client, message):
    log.info(f"About to send message: '{message}'")
    client.messages.create(to=twilio_to,
                           from_=twilio_from,
                           body=message)


def main():
    amazon = AmazonAPI(amazon_access_key, amazon_secret, amazon_associate_tag)
    # Only search the first two pages to not spam the server
    products = amazon.search_n(2, Keywords=search_product_keywords,
                               SearchIndex=search_index, region=search_region)
    twilio_client = Client(twilio_key, twilio_secret)

    try:
        prods = Box.from_json(filename="products.json")
    except FileNotFoundError:
        prods = Box()

    while True:
        new_prods = update(products, prods)
        if new_prods:
            message = format_message(new_prods)
            send_message(twilio_client, message)
        sleep(60)


if __name__ == '__main__':
    main()

So now that possibly huge pure python standard library multifaceted application has turned into fifty lines of code (not counting imports / globals) designed to do nothing but feed my anxiety as efficiently as possible. Luckily it’s self testing to make sure it works, as it will find the result for the Vega Frontier Edition first, so if you chose to use it, make sure you get that text.

2017-08-07 22:09:23,938 - vega             INFO      About to send message: 
'New Prods: Radeon Vega Frontier Edition Liquid Retail-1579.48-
https://www.amazon.com/Radeon-Vega-Frontier-Liquid-Retail/dp/B072XLR2K7?SubscriptionId=AKIAIF3WXFESZ53UZKDQ&tag=codecalamity-20&linkCode=xm2&camp=2025&creative=165953&creativeASIN=B072XLR2K7'

Be warned that this may spam your phone a lot when the product does drop, and that the code doesn’t have a lot of safety checks as-is so may fail and stop reporting. It also doesn’t check if it is available to buy yet, just that the product page for the Vega 64 exists.

If you don’t want to have your own script constantly running, try out sites like nowinstock.net, as they have great options to text or email you when products are available.