Source Code for my talk @ PyRE about REST API with Flask & SQLAlchemy

restapi_talk.py

from flask import Flask, jsonify, abort, request
from flask_login import LoginManager, current_user
import json
from flask_sqlalchemy import SQLAlchemy
from flask_sqlalchemy import orm

db = SQLAlchemy()


class Record(db.Model):
    __tablename__ = "collection"

    index = db.Column(db.Integer, primary_key=True)
    Artist = db.Column(db.Text)
    Title = db.Column(db.Text)
    Label = db.Column(db.Text)
    Released = db.Column(db.Text)

    def as_dict(self):
        return {col.name: getattr(self, col.name) for col in orm.class_mapper(self.__class__).mapped_table.c}

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///record_collection"
db.init_app(app)
login_manager = LoginManager(app)

@login_manager.request_loader
def check_token(request):
    token = request.headers.get('Authorization')
    if token == 'L3T_M3_PA55!':
        return "You_can_pass"  # DON'T TRY THIS AT HOME!
    return None

@app.route("/records", methods=['GET', 'POST', 'PUT', 'DELETE'])
def get_records():
    if request.method == 'POST':
        record = Record(**json.loads(request.data))
        db.session.add(record)
        db.session.commit()
        return jsonify(record.as_dict()), 201
    elif request.method == 'PUT':
        abort(405)
    records = [r.as_dict() for r in Record.query.all()]
    if request.method == 'DELETE':
        if current_user:
            for r in records:
                db.session.delete(r)
            db.session.commit()
            records = [r.as_dict() for r in Record.query.all()]
            return jsonify(records), 200
        else:
            abort(401)
    return jsonify(records), 200


@app.route("/records/<int:index>", methods=['GET', 'POST', 'PUT', 'DELETE'])
def get_record(index):
    if request.method == 'POST':
        abort(405)
    else:
        record = Record.query.filter(Record.index == index).first_or_404()
        if request.method == 'PUT':
            if current_user:
                for k, v in json.loads(request.data).iteritems():
                    setattr(record, k, v)
                db.session.add(record)
                db.session.commit()
            else:
                abort(401)
        elif request.method == 'DELETE':
            if current_user:
                db.session.delete(record)
                db.session.commit()
            else:
                abort(401)
        return jsonify(record.as_dict()), 200

@app.errorhandler(404)
def page_not_found(error):
    return jsonify(
        error="Not Found",
        status_code=404
    ), 404

@app.errorhandler(405)
def method_not_allowed(error):
    return jsonify(
        error="Method Not Allowed",
        status_code=405
    ), 405

@app.errorhandler(401)
def unauthorized(error):
    return jsonify(
        error="Unauthorized access",
        status_code=401
    ), 401

if __name__ == '__main__':
  app.run()