Creating beautiful REST APIs with Flask

Introduction

Presentation located at: pycoder.net

email: brendan.kohler@pycoder.net

twitter: @xnomagichash

Technical Debt: Enemy of Weekends

A summary of Seldera's codebase before we achieved product/market fit:

• Lots of Django
• Many incompatible branches
• Several years of technical debt

What can we do about it?

1. Nuke it from orbit!

2. Maintain the code base

3. Gradually phase old systems out

Phasing old systems out

As your systems scale, your architecture gets more exotic. Planning for change is not premature optimization.

1. Move to service oriented architecture with REST APIs

2. Migrate from Django to Flask for new all new services

3. Move to a schemaless (NoSQL) database

4. Plan for many types of clients

...but building nice APIs is hard and we made a lot of mistakes at first.

Example Endpoint: Creating a User

Simple, right? Let's enumerate the steps...

1. Decode a JSON object containing an email address and password.
2. Validate the email address and password.
3. Hash the password with bcrypt.
4. Insert the new user into the data store.
5. Send a welcome email to the new user.
6. Send an "New User" notification to the system administator.
7. Create a new session for the user.
8. Return a JSON response to the client indicating success or failure.
9. Enforce a rate limit to prevent abuse and/or attacks.

This could wind up being complicated after all.

The Flask Anti-Pattern:
putting everything in the endpoint

@api.route('/user', methods=['POST'])
def create_user():
    data = request.get_json(force=True)
    email    = json.get('email','')
    password = json.get('password','')
    attrs = {}

    if not re.match("^[^@ ]+@[^@ ]+\.[^@ ]+$", email):
        raise ValidationError()
    attrs['email'] = email

    if len(password) < 7:
        raise ValidationError()
    attrs['password'] = bcrypt.hashpw(password, bcrypt.gensalt())
    attrs['id'] = str(db.users.insert(attrs))
    del attrs['password']
    session['id'] = new_user['id']
    notify('admins', 'New User', repr(new_user))
    envelope = Envelope(
        to_addr   = new_user['email'],
        subject   = api.config['WELCOME_EMAIL_SUBJECT'],
        text_body = render_template('emails/new_user.txt'),
        html_body = render_template('emails/new_user.html'))
    def sender():
        envelopes.connstack.push_connection(conn)
        smtp = envelopes.connstack.get_current_connection()
        smtp.send(envelope)
        envelopes.connstack.pop_connection()
    gevent.spawn(sender)
    
    return jsonify({'status': 'OK'})

Flask API "DON'Ts"

1. DON'T use your version control system to version your APIs

   You'll just wind up with a bunch of incompatible code branches that you might not be able to merge.

2. DON'T write endpoints you can't test

   Flask makes it easy to test your endpoints using signals, so you have no excuse.

3. DON'T use lots of extensions

   For any application of sufficient complexity, you will outgrow your extensions. The one exception: when you write your own.

Flask API "DOs"

1. Implement API versions as blueprints

   This way your compatibility-breaking changes are modularized and separated by url.

2. Use signals and mocks for testing

   Signals and mocks make unit testing easy if you modularize your API carefully

3. Use decorators as a code-reuse pattern

   Flask is flexible, so use decorators to abstract out reusable patterns.

4. Use Flask's custom error handler capability

   The one feature that nobody uses, but everyone should. You should almost never use abort in your API endpoints.

A sample application structure

Real Time Monitoring API:

realmon/
    server.py
    api/
        __init__.py
        v1/                  <--- API major version
            __init__.py      <--- Blueprint location
            decorators.py    <--- Decorators
            errors.py        <--- Custom Error Handlers
            endpoints.py     <--- API endpoints
            models/
                __init__.py
                building.py
                user.py      <--- All validators and code
                meter.py          related to the model layer
                env.py
            templates/
                ...
            tests/
                ...

What that endpoint should look like

@api.route('/users', methods=['POST'])
@limit(requests=10, window=60, by="ip")
@email
def create_user():
    data = request.get_json(force=True)
    new_user = user.create(json=data)
    session['uid'] = new_user['id']
    notify('admins', 'New User', repr(new_user))
    return Envelope(
        to_addr   = new_user['email'],
        subject   = api.config['WELCOME_EMAIL_SUBJECT'],
        text_body = render_template('emails/new_user.txt'),
        html_body = render_template('emails/new_user.html'))

*Envelope comes from the excellent email library envelopes.

API Decorators

The Goal: Find common API patterns and abstract them out into decorators

1. Sending Emails

   Often an endpoint's primary output is an email, rather than JSON, but it can be messy to send emails synchronously.

2. Rate Limiting Requests

   It's always a good idea to carefully rate limit your API. It prevents accidental server overloading and malicious behavior.

3. Background Tasks

   There are often tasks API needs to perform that take too long to do synchronously. In those cases it's often better to run the task asychronously and let initiator of the request retrieve the results when the task has finished.

A short refresher on decorators

def decorator(f):
    @functools.wraps(f)
    def wrapped(*args, **kwargs):
        results = f(*args, **kwargs)
        # do something
        return results
    return wrapped

def decorator_args(a, b, c):
    def decorator(f):
        @functools.wraps(f)
        def wrapped(*args, **kwargs):
            results = f(*args, **kwargs)
            # do something
            return results
        return wrapped
    return decorator

The email decorator

def email(f):
    @functools.wraps(f)
    def wrapped(*args, **kwargs):
        envelope = f(*args, **kwargs)
        envelope.from_addr = api.config['SYSTEM_EMAIL']
        def task():
            smtp().send(envelope)
        gevent.spawn(task)
        return jsonify({"status": "OK"})
    return wrapped

Using the email decorator

@api.route('/users', methods=['POST'])
@limit(requests=100, interval=3600, by="ip")
@email
def create_user():
    data = request.get_json(force=True)
    new_user = user.create(json=data, retrieve=True)
    session['id'] = new_user['id']
    notify('admins', 'New User', repr(new_user))
    return Envelope(
           to_addr   = new_user['email'],
           subject   = api.config['WELCOME_EMAIL_SUBJECT'],
           text_body = render_template('emails/new_user.txt'),
           html_body = render_template('emails/new_user.html'))

The limit decorator

First let's start with some definitions:

• requests

  The number of requests that are allowed within a certain time window

• window

  The length of the time window in seconds.

• by

  a function that extracts the key ID that the limit is enforced on.

• group (optional)

  Specifies a group of endpoints that share the same requests pool for rate limiting.

Building the limit decorator

def limit(requests=100, window=60, by="ip", group=None):
    if not callable(by):
        by = { 'ip': lambda: request.headers.remote_addr }[by]

    def decorator(f):
        @functools.wraps(f)
        def wrapped(*args, **kwargs):
            group = group or request.endpoint
            key = ":".join(["rl", group, by()])

            try:
                remaining = requests - int(redis.get(key))
            except (ValueError, TypeError):
                remaining = requests
                redis.set(key, 0)

            ttl = redis.ttl(key)
            if not ttl:
                redis.expire(key, window)
                ttl = window

            g.view_limits = (requests,remaining-1,time()+ttl)

            if remaining > 0:
                redis.incr(key, 1)
                return f(*args, **kwargs)
            else:
                return Response("Too Many Requests", 429)
        return wrapped
    return decorator

X-RateLimit Header Injection

@api.after_request
def inject_rate_limit_headers(response):
    try:
        requests, remaining, reset = map(int, g.view_limits)
    except (AttributeError, ValueError):
        return response
    else:
        h = response.headers
        h.add('X-RateLimit-Remaining', remaining)
        h.add('X-RateLimit-Limit', requests)
        h.add('X-RateLimit-Reset', reset)
        return response

The background decorator

An example:

@api.route('/reports/power/monthly//')
@limit(requests=1, by='ip')
@background
def report_monthly_power(year, month):
    rdata = generate_power_report(year, month)
    return rdata

The background decorator

def background(f):
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        jobid = uuid4().hex
        key = 'job-{0}'.format(jobid)
        skey = 'job-{0}-status'.format(jobid)
        expire_time = 3600
        redis.set(skey, 202)
        redis.expire(skey, expire_time)

        @copy_current_request_context
        def task():
            try:
                data = f(*args, **kwargs)
            except:
                redis.set(skey, 500)
            else:
                redis.set(skey, 200)
                redis.set(key, data)
                redis.expire(key, expire_time)
            redis.expire(skey, expire_time)

        gevent.spawn(task)
        return jsonify({"job": jobid})
    return wrapper

Decorators: A Summary

Find common API patterns and turn them into decorators

Keep the "plumbing" out of your endpoints.

Use libraries like Gevent to run tasks asynchronously

Use rate limiting everywhere

Let errors bubble up from your decorators (and endpoints)

request, session and g make powerful decorators possible

Don't ignore Flask's built-in decorators

Bonus Material: Error Handlers

Here is an exception from models/user.py:

class ValidationError(Exception):
    def __init__(self, field, message):
        self.field = field
        self.message = message

Here is an error handler from errors.py:

@api.errorhandler(user.ValidationError)
def handle_user_validation_error(error):
    response = jsonify({
        'msg': error.message,
        'type': 'validation',
        'field': error.field })
    response.status_code = 400
    return response

Save yourself from doing error handling in your endpoints. Write JSON-friendly error handlers like this.