Don’t repeat yourself (DRY) is a principle of software development aimed at reducing repetition of software patterns, replacing it with abstractions or using data normalization to avoid redundancy.

Decorators are an awesome feature of Python. They allow you to wrap your functions in a so-called decorator to get extra functionality for free. Let’s see how we can use them to reduce duplicate code!

Common examples of decorators are Flask routes:

@app.route('/')
def index():
  return 'Welcome to my website!'

Here the decorator will make sure our route is registered at the provided path, and will call the supplied function when someone sends a request to the root of your website.

Simple decorators

Creating your own decorators is also super simple. Let’s look at a decorator that adds all database model classes we decorate with it to a global array, so we can create database tables on app startup:

DATABASE_MODELS = []

def register(f):
  GLOBAL_ARRAY.append(f)
  return f


@register
class MyModel(DatabaseModel):
  pass


def setup_db():
  for model in DATABASE_MODEL:
    # Setup the models table

These decorators are pretty cool, but what would happen if we could change the function we’re decorating. The possibilities are endless.

Decorators for auditing

Let’s say we have a CRUD API for a peewee model in a Flask app. We want to add some auditing to the create, update and delete methods so we can see who did what. We can use a decorator for this!

We have the following audit model, and we will create a new row in the DB every time audited endpoints are hit.

import functools
from datetime import datetime

from flask import request
from peewee import *
from playhouse.postgres_ext import *

class AuditModel(Model):
  id = AutoField()
  user = TextField()
  action = TextField()
  timestamp = DateTimeField()
  metadata = BinaryJSONField()

  abort = False

  class Meta:
    db_table = 'audits'

We can now add a staticmethod to this class that will be our decorator:

@staticmethod
def log(f=None):
  def deco(f):
    @functools.wraps(f)
    def func(*args, **kwargs):
      audit = AuditModel()
      user = request.headers.get("X-Auth-User")
      if user is None:
        return 'UNAUTHORIZED. Please provide X-Auth-User header.', 403
      audit.user = user
      audit.action = f'{f.__module__}.{f.__name__}'
      audit.timestamp = datetime.utcnow()
      if 'audit' in f.__code__.co_varnames:
        kwargs['audit'] = audit
      result = f(*args, **kwargs)
      if not audit.abort:
        audit.save(force_insert=True)
      return result
    return func

  if f and callable(f):
    return deco(f)

  return deco

Let’s break down what’s going on here: By using functools.wraps, our internal func function will be called with the same arguments that would be passed in to the function we’re decorating. This allows us to do things with these arguments.

Our first step is to create a new AuditModel instance and check if we have a user. Here it’s just a simple header, but this could be more sophisticated.

Once we’re sure we have a logged in user, we add the current date-time, and set the action to the module and function name of the decorated function. This will end up something like mymodule.myfile.myfunction.

Next up we check if the function we’re decorating has an audit argument. If so, we add our model instance to the kwargs. This allows the decorated function to set or change the metadata.

We then save the result of the decorated function (the actual flask response) and save the audit model. Note that we have a check on abort here. This will allow route handlers to cancel the audit saving in case of, for example, when no action is taken.

After saving, we return the captured response so that our response gets sent like normal. In a route handler this would look like this:

@app.route('/item/<int:item>', methods=['DELETE'])
@AuditModel.log
def clear_all(item, audit):
  audit.metadata = { 'item_id': item }

  # DELETE item

  return '', 204

So what happens when we send a DELETE request to /item/12 is:

1. Auth is checked.
2. The base AuditModel instance is created.
3. Because the clear_all function accepts an audit parameter, the model is passed into the function.
4. The metadata on the audit is set.
5. A response is returned from the decorated function.
6. The audit model is saved to the database.
7. The response is returned to the end user.

Awesome, right?

Prefetching with decorators

Next up we also don’t want to repeat the database lookup based on the item route param every time. Let’s create another decorator! This one can be added to the Item model class:

@staticmethod
def prefetch(f=None):
  def deco(f):
    @functools.wraps(f)
    def func(*args, **kwargs):
      try:
        item = Item.get(
          Item.id == kwargs.pop('item')
        )
        return f(item, *args, **kwargs)
      except Item.DoesNotExist:
        return 'Item not found', 404
    return func

  if f and callable(f):
    return deco(f)

  return deco

This decorator will take item out of the kwargs (this will be the URL parameter Flask already added for us) and attempt to get an Item from the database. If none are found, a 404 is returned. In our Flask route, this would look like this:

@app.route('/item/<int:item>', methods=['DELETE'])
@Item.prefetch
@AuditModel.log
def clear_all(item, audit):
  audit.metadata = { 'item_id': item.id }

  print(item) # Item instance. Not an int anymore
  # DELETE item

  return '', 204

Here we can combine both decorators together to get auditing and have our Item pre-fetched from the database before our function is called.

This allows us to create our decorators once, and put them on top of all route handlers that need pre-fetching, auditing or both. Way to DRY!

Where to go from here

Thank you for taking the time to read this story! Feel free to leave your comments and ask questions!

I work as a Data Engineer at Wehkamp.nl, one of the biggest e-commerce companies of the Netherlands. This article is part of our Tech Blog, check it out & subscribe. Looking for a great job? Check our job offers.