A not-so-simple web service that tries to illustrate HATEOAS (using handcrafted json+hal) and stateless authorization (using JWT)

app.py

# To assure reproducability, the given Redis database (db=0)
# !!! is CLEARED !!! each time the application starts
#
# $ export JWT_SECRET=something
# $ python3 -m pip install flask pyjwt redis
# $ export REDIS_PASSWORD=verycomplex
# $ export REDIS_PORT=6379
# $ export REDIS_HOST=myhost.com
from os import getenv
from flask import Flask, g
from flask import request
from flask import make_response
from json import loads, dumps
app = Flask(__name__)

from jwt import decode
from jwt.exceptions import DecodeError, ExpiredSignatureError
SECRET = getenv("JWT_SECRET")

from redis import StrictRedis
PORT = getenv("REDIS_PORT")
HOST = getenv("REDIS_HOST")
PASS = getenv("REDIS_PASSWORD")
db = StrictRedis(host=HOST, port=int(PORT), password=PASS, db=0, decode_responses=True)
db.flushdb() # clear all data from the current db
db.hset("stock", "carrot", 10)
db.hset("price", "carrot", 5)
db.hset("stock", "apple", 20)
db.hset("price", "apple", 3)
db.hset("stock", "pineapple", 0)
db.hset("price", "pineapple", 10)

from dao import Price, Stock, Orders
price = Price(db)
stock = Stock(db)
orders = Orders(db)

@app.before_request
def authorize():
  g.uid = None
  g.role = None
  g.error = None
  if "Authorization" not in request.headers:
    g.error = "No Authorization header"
    return

  header = request.headers["Authorization"]
  if "Bearer" not in header:
    g.error = "Only Bearer tokens allowed"
    return

  token = header[len("Bearer "):]
  try:
    g.payload = decode(token, SECRET, audience="grocery.shop", algorithms=["HS256"], options={
     "require": ["exp", "aud"]
    })
  except ExpiredSignatureError:
    g.error = "Token expired"
    return
  except DecodeError:
    g.error = "Invalid token"
    return

  g.role = g.payload.get("role")
  g.uid = g.payload.get("uid")

@app.route("/orders", methods=["GET"])
def list_user_orders():
  if g.error:
    return g.error, 403
  if g.uid is None:
    return "Could not identify user", 403
  if g.role is None:
    return "Could not identify role", 403

  result = []
  if g.role == "customer":
    oids = orders.for_user(g.uid)
  if g.role == "admin":
    oids = orders.all()

  for oid in oids:
    order, total, owner, state = orders[oid]
    links = {}
    links["self"] = { "href": f"/orders/{oid}", "method": "GET" }
    if state == "unpaid":
      links["pay"] = { "href": f"/orders/{oid}", "method": "PATCH" }
      links["cancel"] = { "href": f"/orders/{oid}", "method": "DELETE" }
    if state == "paid":
      links["track"] = { "href": f"/orders/{oid}/shipping", "method": "GET" }
    order["_links"] = links
    result.append(order)
    
  links = {}
  links["self"] = { "href": "/orders", "method": "GET" }
  links["create"] = { "href": "/orders", "method": "POST" }
  response = make_response(dumps({
    "orders": result,
    "_links": links
  }, indent=None), 200)
  
  response.headers["Content-Type"] = "application/json"
  return response
  
@app.route("/orders", methods=["POST"])
def create_order():
  if g.error:
    return g.error, 403
  if g.uid is None:
    return "Could not identify user", 403
  if g.role is None:
    return "Could not identify role", 403

  messages = []
  cart = {} # default value, can be loaded from cookies
  cookies = request.headers.get("Cookie", "")
  for cookie in cookies.split(";"): # foo=1; cart={}; bar=true; baz=abc
    if cookie.startswith("cart="):
      cart = loads(cookie[len("cart="):]) # assume that it is JSON
  
  products = cart.keys()
  order = {}
  for product in products:
    if product not in stock:
      messages.append(f"{product} not in stock!")
      continue
    
    if stock[product] >= cart[product]:
      units = cart[product]
      cost = units * price[product]
      order[product] = (units, cost)
      stock[product] = stock[product] - units
    else:
      messages.append(f"Not enough {product} in stock: {stock[product]} vs. {cart[product]} needed. Skipping the item...")
      continue

  items = order.keys()
  if len(items) == 0:
    messages.append("The cart from the cookie header has no items...")
    return dumps(messages), 400

  total = 0
  for item in items:
    units, cost = order[item]
    total += cost
  
  uid = g.uid
  oid = orders.create(uid, order, total)

  links = {}
  links["pay"] = { "href": f"/orders/{oid}", "method": "PATCH" }
  links["self"] = { "href": f"/orders/{oid}", "method": "GET" }
  links["cancel"] = { "href": f"/orders/{oid}", "method": "DELETE" }

  response = make_response(dumps({
    "oid": oid,
    "messages": messages,
    "_links": links
  }, indent=None), 200)
  response.headers["Set-Cookie"] = "cart={}; HttpOnly"
  response.headers["Location"] = f"/orders/{oid}"
  response.headers["Content-Type"] = "application/json"
  return response

@app.route("/orders/<oid>", methods=["GET"])
def get_order(oid):
  if g.error:
    return g.error, 403
  if g.uid is None:
    return "Could not identify user", 403
  if g.role is None:
    return "Could not identify role", 403

  result = orders[oid]
  if result:
    order, total, owner, state = result
    if g.uid != owner and g.role != "admin":
      return "Not an owner", 403

    links = {}
    links["self"] = { "href": f"/orders/{oid}", "method": "GET" }
    if g.role == "admin" or state != "paid":
      links["cancel"] = { "href": f"/orders/{oid}", "method": "DELETE" }
    if g.uid == owner and state == "unpaid":
      links["pay"] = { "href": f"/orders/{oid}", "method": "PATCH" }
    if g.uid == owner and state == "paid":
      links["track"] = { "href": f"/orders/{oid}/shipping", "method": "GET" }

    response = make_response(dumps({
      "oid": oid,
      "order": order,
      "total": total,
      "state": state,
      "_links": links
    }, indent=None), 200)
    response.headers["Content-Type"] = "application/json"
    return response

  return "Order not found", 404

@app.route("/orders/<oid>/shipping", methods=["GET"])
def track_order(oid):
  if g.error:
    return g.error, 403
  if g.uid is None:
    return "Could not identify user", 403
  if g.role is None:
    return "Could not identify role", 403

  result = orders[oid]
  if result:
    order, total, owner, state = result
    if g.uid != owner and g.role != "admin":
      return "Not an owner", 403

    if state != "paid":
      return "No tracking info", 404

    links = {}
    links["self"] = { "href": f"/orders/{oid}/shipping", "method": "GET" }

    response = make_response(dumps({
      "info": "Waiting for dispatching",
      "_links": links
    }, indent=None), 200)
    response.headers["Content-Type"] = "application/json"
    return response

  return "Order not found", 404

@app.route("/orders/<oid>", methods=["DELETE"])
def cancel_order(oid):
  if g.error:
    return g.error, 403
  if g.uid is None:
    return "Could not identify user", 403
  if g.role is None:
    return "Could not identify role", 403

  result = orders[oid]
  if result:
    order, total, owner, state = result
    if g.uid == owner and state == "paid":
      return "Can't cancel paid orders", 400

    if g.uid != owner and g.role != "admin":
      return "Not an owner", 403

    orders.cancel(oid)

    links = {}
    links["orders"] = { "href": f"/orders", "method": "GET" }

    response = make_response(dumps({
      "_links": links
    }, indent=None), 200)
    response.headers["Content-Type"] = "application/json"
    return response

  return "Order not found", 404


@app.route("/orders/<oid>", methods=["PATCH"])
def pay_order(oid):
  if g.error:
    return g.error, 403

  result = orders[oid]
  if result:
    order, total, owner, state = result
    if state != "unpaid":
      return "You can only pay for unpaid orders", 400

    payment = g.payload.get("payment")
    if payment:
      orders.set_state(oid, "paid")
    else:
      return "No valid payment identifier", 400

    links = {}
    links["self"] = { "href": f"/orders/{oid}", "method": "GET" }
    links["track"] = { "href": f"/orders/{oid}/shipping", "method": "GET" }

    response = make_response(dumps({
      "_links": links
    }, indent=None), 200)
    response.headers["Content-Type"] = "application/json"
    return response

  return "Order not found", 404
  
@app.route("/")
def index():
  if g.uid:
    links = {}
    links["orders"] = { "href": f"/orders", "method": "GET" }
    response = make_response(dumps({
      "_links": links
    }, indent=None), 200)
    response.headers["Content-Type"] = "application/json"
    return response
  return "You have to obtain a valid authorization token to use this API", 403

# $ python3 app.py
app.run() # point POSTMAN to http://127.0.0.1:5000/