REST API Coding Interview Questions (With Python and Flask Answers)

2. Add authentication to a route

Question: Protect one endpoint so only authenticated callers reach it.

from functools import wraps
from flask import request, jsonify

TOKENS = {'tok_live_abc': 'user_1'}


def require_auth(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        header = request.headers.get('Authorization', '')
        if not header.startswith('Bearer '):
            return jsonify({'error': {'code': 'unauthenticated',
                                      'message': 'Missing bearer token.'}}), 401
        user = TOKENS.get(header.split(' ', 1)[1])
        if not user:
            return jsonify({'error': {'code': 'unauthenticated',
                                      'message': 'Invalid token.'}}), 401
        request.user_id = user
        return f(*args, **kwargs)
    return wrapper


@app.route('/v1/me', methods=['GET'])
@require_auth
def me():
    return jsonify({'user_id': request.user_id}), 200

Say the 401 versus 403 distinction out loud, because it is a common follow up. 401 means we do not know who you are. 403 means we do, and you still may not do this.

Two more points score well. A token belongs in the Authorization header, never in a query string, because query strings land in logs. Real systems verify a signed token rather than reading a dictionary. Name JWT verification or an introspection call as the production version.

3. Support more than one API version

Question: Serve version 1 and version 2 of an endpoint at the same time.

from flask import Blueprint

v1 = Blueprint('v1', __name__, url_prefix='/v1')
v2 = Blueprint('v2', __name__, url_prefix='/v2')


@v1.route('/users/<int:user_id>')
def get_user_v1(user_id):
    return jsonify({'id': user_id, 'name': 'Ada Lovelace'}), 200


@v2.route('/users/<int:user_id>')
def get_user_v2(user_id):
    # v2 splits the single name field into two.
    return jsonify({'id': user_id,
                    'first_name': 'Ada',
                    'last_name': 'Lovelace'}), 200


app.register_blueprint(v1)
app.register_blueprint(v2)

Blueprints keep each version's handlers separate, so a change to v2 cannot break v1 by accident.

The interviewer usually asks a second question here: when do you cut a new version? Only when a change would break a working client. Splitting name into two fields is breaking, because a client reading name now gets nothing. Adding a nickname field is not breaking, so it belongs in v1.

4. Return proper errors and status codes

Question: Handle errors so clients can react to them programmatically.

class ApiError(Exception):
    def __init__(self, code, message, status):
        self.code, self.message, self.status = code, message, status


@app.errorhandler(ApiError)
def handle_api_error(e):
    return jsonify({'error': {'code': e.code, 'message': e.message}}), e.status


@app.errorhandler(404)
def handle_404(e):
    return jsonify({'error': {'code': 'not_found',
                              'message': 'No such endpoint.'}}), 404


@app.errorhandler(Exception)
def handle_unexpected(e):
    app.logger.exception('unhandled error')
    return jsonify({'error': {'code': 'internal_error',
                              'message': 'Something went wrong.'}}), 500


@app.route('/v1/orders/<int:order_id>/cancel', methods=['POST'])
def cancel_order(order_id):
    order = orders.get(order_id)
    if not order:
        raise ApiError('not_found', 'No order with that id.', 404)
    if order['status'] == 'shipped':
        raise ApiError('already_shipped', 'This order has shipped.', 409)
    order['status'] = 'cancelled'
    return jsonify(order), 200

One error envelope for the whole API means a client writes its error handling once. The code is the contract and must stay stable. The message is for humans and can be reworded.

Note the 409 on the shipped order. The request was well formed and the caller was allowed, but the resource is in the wrong state. Reaching for 400 there is the common mistake.

Never leak an exception message to the caller. Log the detail, return a generic sentence.

5. Add rate limiting

Question: Stop a single client from calling an endpoint too often.

import time
from collections import defaultdict
from flask import request, jsonify

LIMIT, WINDOW = 100, 60
hits = defaultdict(list)


def rate_limit(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        key = request.headers.get('Authorization') or request.remote_addr
        now = time.time()
        hits[key] = [t for t in hits[key] if now - t < WINDOW]
        remaining = LIMIT - len(hits[key])
        if remaining <= 0:
            retry_after = int(WINDOW - (now - hits[key][0])) + 1
            return jsonify({'error': {'code': 'rate_limited',
                                      'message': 'Too many requests.'}}), 429, {
                'Retry-After': str(retry_after),
                'X-RateLimit-Limit': str(LIMIT),
                'X-RateLimit-Remaining': '0'}
        hits[key].append(now)
        response = f(*args, **kwargs)
        return response
    return wrapper

The headers are the part most candidates skip, and they are the part that matters. A client that receives Retry-After waits. A client that receives a bare error retries immediately and makes the problem worse.

Expect the scaling follow up. This counter lives in one process, so it breaks the moment you run two servers. The production answer is a shared store such as Redis, with the trade off that Redis becomes a dependency on every request.

6. Paginate a list endpoint

Question: Return a large collection in pages.

import base64
import json


def encode_cursor(last_id):
    return base64.urlsafe_b64encode(json.dumps({'id': last_id}).encode()).decode()


def decode_cursor(cursor):
    try:
        return json.loads(base64.urlsafe_b64decode(cursor.encode()))['id']
    except Exception:
        raise ApiError('invalid_cursor', 'The cursor is not valid.', 400)


@app.route('/v1/posts', methods=['GET'])
def list_posts():
    limit = min(int(request.args.get('limit', 20)), 100)
    cursor = request.args.get('cursor')
    rows = sorted(posts.values(), key=lambda p: p['id'], reverse=True)
    if cursor:
        last_id = decode_cursor(cursor)
        rows = [p for p in rows if p['id'] < last_id]
    page = rows[:limit]
    return jsonify({
        'data': page,
        'next_cursor': encode_cursor(page[-1]['id']) if len(rows) > limit else None,
        'has_more': len(rows) > limit,
    }), 200

Two things score here. The limit is capped, so a caller cannot ask for a million rows. And the cursor is encoded, which signals to clients that it is opaque and not something to construct by hand.

Be ready to compare the two approaches. Offset pagination is simpler and supports jumping to any page, but the database reads and discards the skipped rows, so deep pages get slow. It also repeats and skips items when new rows arrive mid read. Cursor pagination stays fast at any depth and stays correct while the list grows.