Flask/REST API
< Flask
main.py
[edit | edit source]# References:
# https://blog.miguelgrinberg.com/post/designing-a-restful-api-with-python-and-flask
from flask import abort
from flask import Flask
from flask import jsonify
from flask import make_response
from flask import request
import database
app = Flask(__name__, root_path='.')
@app.route('/<path:path>')
def catch_all(path):
abort(404)
@app.route('/users', methods=['GET'])
def get_users():
rows = database.select_users()
if rows == None:
abort(404)
return jsonify([dict(row) for row in rows])
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
row = database.select_user(user_id)
if row == None:
abort(404)
return jsonify(dict(row))
@app.route('/users', methods=['POST'])
def insert_user():
if not request.json or not "user" in request.json:
abort(400)
user = request.json['user'].strip()
if len(user) == 0:
abort(400)
row = database.insert_user(user)
if row == None:
abort(400)
return jsonify(dict(row))
@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
if not request.json or not "user" in request.json:
abort(400)
user = request.json['user'].strip()
if len(user) == 0:
abort(400)
row = database.update_user(user_id, user)
if row == None:
abort(400)
return jsonify(dict(row))
@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
database.delete_user(user_id)
return jsonify({'result': True})
@app.errorhandler(400)
def bad_request(error):
return make_response(jsonify({'error': 'Bad request'}), 400)
@app.errorhandler(404)
def not_found(error):
return make_response(jsonify({'error': 'Not found'}), 404)
@app.errorhandler(405)
def method_not_allowed(error):
return make_response(jsonify({'error': 'Method not allowed'}), 405)
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port='5000')
database.py
[edit | edit source]import sqlite3
DATABASE = "users.db"
def check_database(connection):
try:
cursor = connection.cursor()
sql = """
SELECT name FROM sqlite_master
WHERE type='table' AND name='Users';
"""
cursor.execute(sql)
row = cursor.fetchone()
if row == None:
create_table(connection)
except:
print("Unable to select from sqlite_master")
raise
def create_table(connection):
try:
cursor = connection.cursor()
sql = "DROP TABLE IF EXISTS Users;"
cursor.execute(sql)
sql = """
CREATE TABLE Users(
UserID INTEGER PRIMARY KEY AUTOINCREMENT,
User TEXT NOT NULL UNIQUE
);
"""
cursor.execute(sql)
connection.commit()
except Exception as exception:
print("Unable to execute %s" % sql)
print(exception)
def insert_user(user):
try:
connection = sqlite3.connect(DATABASE)
check_database(connection)
except:
print("Unable to connect to %s" % DATABASE)
raise
try:
cursor = connection.cursor()
sql = f"INSERT INTO Users(user) VALUES('{user}');"
cursor.execute(sql)
sql = "SELECT last_insert_rowid()"
cursor.execute(sql)
row = cursor.fetchone()
connection.commit()
return select_user(row[0])
except Exception as exception:
print("Unable to execute %s" % sql)
print(exception)
finally:
connection.close()
def select_users():
try:
connection = sqlite3.connect(DATABASE)
check_database(connection)
except:
return("Unable to connect to %s" % DATABASE)
try:
connection.row_factory = sqlite3.Row
cursor = connection.cursor()
sql = """
SELECT UserID, User FROM Users;
"""
cursor.execute(sql)
rows = cursor.fetchall()
return rows
except Exception as exception:
print("Error processing %s" % sql)
print(exception)
finally:
connection.close()
def select_user(user_id):
try:
connection = sqlite3.connect(DATABASE)
check_database(connection)
except:
return("Unable to connect to %s" % DATABASE)
try:
connection.row_factory = sqlite3.Row
cursor = connection.cursor()
sql = f"""
SELECT UserID, User FROM Users
WHERE UserID = {user_id};
"""
cursor.execute(sql)
row = cursor.fetchone()
return row
except Exception as exception:
print("Error processing %s" % sql)
print(exception)
finally:
connection.close()
def update_user(user_id, user):
try:
connection = sqlite3.connect(DATABASE)
check_database(connection)
except:
print("Unable to connect to %s" % DATABASE)
raise
try:
cursor = connection.cursor()
sql = f"UPDATE Users SET User = '{user}' WHERE UserID = {user_id};"
cursor.execute(sql)
connection.commit()
return select_user(user_id)
except Exception as exception:
print("Unable to execute %s" % sql)
print(exception)
finally:
connection.close()
def delete_user(user_id):
try:
connection = sqlite3.connect(DATABASE)
check_database(connection)
except:
print("Unable to connect to %s" % DATABASE)
raise
try:
cursor = connection.cursor()
sql = f"DELETE FROM Users WHERE UserID = {user_id};"
cursor.execute(sql)
connection.commit()
except Exception as exception:
print("Unable to execute %s" % sql)
print(exception)
finally:
connection.close()
Try It
[edit | edit source]Copy and paste the code above into one of the following free online development environments or use your own REST API compiler / interpreter / IDE.
Test the API using cURL commands:
curl -i http://localhost:3000/users
curl -i -H "Content-Type: application/json" -X POST -d '{"user":"Alice"}' http://localhost:3000/users
curl -i -H "Content-Type: application/json" -X POST -d '{"user":"Bob"}' http://localhost:3000/users
curl -i http://localhost:3000/users
curl -i http://localhost:3000/users/1
curl -i -H "Content-Type: application/json" -X PUT -d '{"user":"Carol"}' http://localhost:3000/users/1
curl -i http://localhost:3000/users
curl -i -X DELETE http://localhost:3000/users/2
curl -i http://localhost:3000/users