RETRIEVE a representation of the addressed member of the collection
REPLACE the addressed member of the collection.
Method not allowed.
DELETE the addressed member of the collection.
What do we not care for this evening
Stability & Testing
Long-Term maintainability
Edge Cases
Operations, Caching & Deployment
My Vinyl Collection API
http://flask.pocoo.org/
Hello API!
from flask import Flask
app = Flask(__name__)
if __name__ == '__main__':
app.run()
$ python myvinylcollectionapi.py
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
Hello API!
from flask import Flask
app = Flask(__name__)
@app.route("/")
def hello():
return "Hello PyRE!"
if __name__ == '__main__':
app.run()
Hello API!
from flask import Flask, jsonify
app = Flask(__name__)
@app.route("/")
def hello():
return jsonify(data="Hello PyRE!")
if __name__ == '__main__':
app.run()
HTTP GET Methods
URL
GET
http://api.myvinylcollection.com/records
LIST of records in collection
http://api.myvinylcollection.com/records/1
RETRIEVE a representation of the addressed member of the collection
from flask import Flask, jsonify, abort
app = Flask(__name__)
RECORDS = [
{
'id': 0,
'artist': "Queen",
'title': "A Night At The Opera",
'year': "1975",
'label': "EMI"
},
{
'id': 1,
'artist': "Pink Floyd",
'title': "The Dark Side Of The Moon",
'year': "1989",
'label': "EMI"
},
...
]
$ curl -X GET localhost:5000/records
[
{
"artist": "Queen",
"id": 0,
"label": "EMI",
"title": "A Night At The Opera",
"year": "1975"
},
{
"artist": "Pink Floyd",
"id": 1,
"label": "EMI",
"title": "The Dark Side Of The Moon",
"year": "1989"
}
]
HTTP GET Method
$ curl -X GET localhost:5000/records/1
{
"artist": "Pink Floyd",
"id": 1,
"label": "EMI",
"title": "The Dark Side Of The Moon",
"year": "1989"
}
$ curl -X GET localhost:5000/records/5
404 Not Found
Not Found
The requested URL was not found on the server.
If you entered the URL manually please check
your spelling and try again.
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, nullable=False)
Title = db.Column(db.Text, nullable=False)
Label = db.Column(db.Text)
Released = db.Column(db.Text)
def as_dict(self):
columns = orm.class_mapper(self.__class__).mapped_table.c
return {
col.name: getattr(self, col.name)
for col in columns
}
Query
>>> # .all() return a list
>>> all_records = Record.query.all()
>>> len(all_records)
80
>>> # .first() return the first item that matches
>>> record = Record.query.filter(Record.index == 9).first()
>>> record.Title
"Back In Black"
>>> record.Artist
"AC/DC"
>>> record.Released
"1980"
>>> # .filter_by() is a shortcut
>>> record = Record.query.filter_by(index == 6).first()
>>> record.Title
"Hotel California"
from flask import Flask, jsonify
from model import db, Record
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///record_collection.db"
db.init_app(app)
@app.route("/records")
def get_records():
records = [r.as_dict() for r in Record.query.all()]
return jsonify(records)
@app.route("/records/<int:index>")
def get_record(index):
record = Record.query.filter(Record.index == index).first_or_404()
return jsonify(record.as_dict())
$ curl -X GET localhost:5000/records
[
{
"Artist": "The Police",
"index": 0,
"Title": "Reggatta De Blanc"
},
{
"Artist": "The Beatles",
"index": 1,
"Title": "Abbey Road"
},
...
]
$ curl -X GET localhost:5000/records/1
{
"Artist": "The Beatles",
"index": 1,
"Title": "Abbey Road"
}
from flask import Flask, jsonify, abort, request
...
@app.route("/records/<int:index>", methods=['GET', 'POST'])
def get_record(index):
if request.method == 'POST':
abort(405)
record = Record.query.filter(Record.index == index).first_or_404()
return jsonify(record.as_dict())
$ curl -X POST localhost:5000/records/1
{
"error": "Method Not Allowed",
"status_code": 405
}
Insert into database
>>> # .add() insert a record
>>> db.session.add(record)
>>> # changes won't be saved until committed!
>>> db.session.commit()
Adding a record to my collection
@app.route("/records", methods=['GET', 'POST'])
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
records = [r.as_dict() for r in Record.query.all()]
return jsonify(records)
@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':
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
DELETE on Resource
@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':
for k, v in json.loads(request.data).iteritems():
setattr(record, k, v)
db.session.add(record)
db.session.commit()
elif request.method == 'DELETE':
db.session.delete(record)
db.session.commit()
return jsonify(record.as_dict()), 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
Securing our API - Collection
@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
