There's no real code just yet. But this series of videos will go indepth so this might be a great moment to grab a coffee!

Functions can accept functions as input.

def apply(func, a, b):
    return func(a, b)

def add(a, b):
    return a + b

def sub(a, b):
    return a - b 

apply(add, 1, 2), apply(sub, 1, 2)

But they can also return functions as output.

def power(n):
    def func(number):
        return number**n
    return func

pow2 = power(2)
pow3 = power(3)

pow2(3), pow3(3)

Let's combine the ideas from the previous video to create a function that accepts a function as input but also returns a function as output.

import time 
import random

def stopwatch(f):
    def func():
        tic = time.time()
        result = f()
        print(f"this function took: {time.time() - tic}")
        return result 
    return func

def sleep_random(s):
    t = s + random.random()
    time.sleep(t)
    return "Done!"

timed_sleep = stopwatch(sleep_random)

You now have two functions that are similar, but also different.

sleep_random()
timed_sleep()

If you're not familiar with args/kwargs you should check out the course on this topic here. Assuming you're familiar, we'll now make the functionality-adding function from the previous video better.

import time 
import random

def stopwatch(f):
    def func(*args, **kwargs):
        tic = time.time()
        result = f(*args, **kwargs)
        print(f"this function took: {time.time() - tic}")
        return result 
    return func

def sleep_random(s):
    t = s + random.random()
    time.sleep(t)
    return "Done"

timed_sleep = stopwatch(sleep_random)

Again, now have two functions that are similar, but they now both allow for an input.

sleep_random(s=2)
timed_sleep(s=2)

We can also use the decorator syntax to achieve the same behavior.

import time 
import random

def stopwatch(f):
    def func(*args, **kwargs):
        tic = time.time()
        result = f(*args, **kwargs)
        print(f"this function took: {time.time() - tic}")
        return result 
    return func

@stopwatch
def sleep_random(s):
    t = s + random.random()
    time.sleep(t)
    return f"Done"

Note that we now no longer have two functions, both one!

sleep_random(1)
sleep_random(2)
sleep_random(3)

Let's make sure that the docstring is kept intact by using @wraps.

import time 
import random

def stopwatch(f):
    def func(*args, **kwargs):
        tic = time.time()
        result = f(*args, **kwargs)
        print(f"this function took: {time.time() - tic}")
        return result 
    return func

@stopwatch
def sleep_random(s):
    """This function sleeps at least for `s` seconds."""
    return time.sleep(s + random.random())

timed_sleep = stopwatch(sleep_random)

Note that we now no longer have two functions, but one!

help(sleep_random)

To ephesize that you can stack decorators, let's apply two new ones.

import time 
import random
from functools import wraps

def print_call1(f):
    @wraps(f)
    def func(*args, **kwargs):
        print(f"print-call 1 args: {args}")
        result = f(*args, **kwargs)
        return result 
    return func


def print_call2(f):
    @wraps(f)
    def func(*args, **kwargs):
        print(f"print-call 2 args: {args}")
        result = f(*args, **kwargs)
        return result 
    return func

@print_call2
@print_call1
@print_call2
@print_call1
def sleep_random(s):
    """This function sleeps at least for `s` seconds."""
    return time.sleep(s + random.random()/100)

sleep_random(1.5)

We can also have decorators that accept inputs. This makes for even more expressiveness.

import time 
import random
from functools import wraps

def loggg(show_name=True, show_time=True):
    def stopwatch(f):
        @wraps(f)
        def func(*args, **kwargs):
            tic = time.time()
            result = f(*args, **kwargs)
            result = "call"
            if show_name:
                result = f"{result} {f.__name__}"
            if show_time:
                result = f"{result} time:{time.time() - tic}"
            return result 
        return func
    return stopwatch

@loggg(show_name=False, show_time=True)
def sleep_random(s):
    """This function sleeps at least for `s` seconds."""
    return time.sleep(s + random.random()/100)

sleep_random(1)

To make inputs optional, we need to do something nitty-gritty.

import time 
import random
from functools import wraps

def loggg(func_in=None, *, show_name=True, show_time=True):
    def stopwatch(f):
        @wraps(f)
        def func(*args, **kwargs):
            tic = time.time()
            result = f(*args, **kwargs)
            result = "call"
            if show_name:
                result = f"{result} {f.__name__}"
            if show_time:
                result = f"{result} time:{time.time() - tic}"
            return result 
        return func

    # This is where the "magic" happens.
    if func_in is None:
        return stopwatch
    else:
        return stopwatch(func_in)

@loggg
def sleep_random(s):
    """This function sleeps at least for `s` seconds."""
    return time.sleep(s + random.random()/100)

sleep_random(1)

Feel free to review this topic again over a period of time. It's very normal if the entire abstraction doesn't 100% "click" right away.

If you're interested in more information on the pandas example below, make sure you watch this course.

The pandas-logger decorator is defined below.

from functools import wraps
import datetime as dt

def log_step(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        tic = dt.datetime.now()
        result = func(*args, **kwargs)
        time_taken = str(dt.datetime.now() - tic)
        print(f"just ran step {func.__name__} shape={result.shape} took {time_taken}s")
        return result
    return wrapper

And you can see it being applied here;

import pandas as pd 

df = pd.read_csv('https://calmcode.io/datasets/bigmac.csv')

@log_step
def start_pipeline(dataf):
    return dataf.copy() 

@log_step
def set_dtypes(dataf):
    return (dataf
            .assign(date=lambda d: pd.to_datetime(d['date']))
            .sort_values(['currency_code', 'date']))

@log_step
def remove_outliers(dataf, min_row_country=32):
    countries = (dataf
                .groupby('currency_code')
                .agg(n=('name', 'count'))
                .loc[lambda d: d['n'] >= min_row_country]
                .index)
    return (dataf
            .loc[lambda d: d['currency_code'].isin(countries)])

df_new = (df
  .pipe(start_pipeline)
  .pipe(set_dtypes)
  .pipe(remove_outliers, min_row_country=20))

Another example of a useful decorator can be found in the retry package.

from retry import retry

import logging
logging.basicConfig()

@retry(ValueError, tries=5, delay=0.5)
def randomly_fails(p=0.5):
    if random.random() < p:
        raise ValueError("no bueno!")
    return "Done!"

randomly_fails()

There's plenty of other useful decorators. There's the lru-cache but there's also the multifile decorator that we mention in the video.