-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
121 lines (86 loc) · 3.42 KB
/
main.py
File metadata and controls
121 lines (86 loc) · 3.42 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
from flask import Flask, current_app, request
from flask_sqlalchemy import SQLAlchemy
import flask
import json
db = SQLAlchemy()
class Folder(db.Model):
__tablename__ = 'folders'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), unique=True, nullable=False)
def __repr__(self):
return '<Folder %r>' % self.name
from sqlalchemy.ext.declarative import DeclarativeMeta
class AlchemyEncoder(json.JSONEncoder):
excludes = ["query", "query_class", "registry"]
def default(self, obj):
if isinstance(obj.__class__, DeclarativeMeta):
# an SQLAlchemy class
fields = {}
for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata' and x not in self.excludes]:
data = obj.__getattribute__(field)
try:
json.dumps(data) # this will fail on non-encodable values, like other classes
fields[field] = data
except TypeError:
fields[field] = None
# a json-encodable dict
return fields
return json.JSONEncoder.default(self, obj)
def create_app():
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite://///tmp/test.db"
db.init_app(app)
with app.app_context():
db.create_all()
@app.errorhandler(404)
def page_not_found(error):
return {"status":"Resource not found, are you sure e.g. the endpoint and id are correct"}, 404
@app.errorhandler(400)
def bad_request(error):
return {"status":"You have made an invalid request , check your paramaters and endpoint"}, 400
@app.errorhandler(500)
def internal_server_error(error):
return {"status":"Unhandeled interal error, please report"}, 500
def render(f):
return json.dumps(f ,cls=AlchemyEncoder)
@app.route("/", methods=["GET"])
def index():
return "<h2>Test API</h2>"
@app.route("/api/v1/folders", methods=["GET"])
def query_all():
folders = db.paginate(db.select(Folder))
print(folders.items)
response = {"status": folders.items}
return render(response)
@app.route("/api/v1/folders/<int:folder_id>", methods=["GET"])
def query(folder_id):
f = db.get_or_404(Folder, folder_id) # bug, validate int
response = {"status": f}
return render(response)
@app.route("/api/v1/folders/<int:folder_id>", methods=["DELETE"])
def delete(folder_id):
f = db.get_or_404(Folder, folder_id) # bug, validate int
db.session.delete(f)
db.session.commit()
response = {"status": f}
return render(response)
@app.route("/api/v1/folders", methods=["POST", "PUT"])
def server_request():
if flask.request.method == 'POST':
payload = request.json
f = Folder(name=payload["folder_name"])
db.session.add(f)
elif flask.request.method == 'PUT':
payload = request.json
folder_name = payload.get("folder_name")
id = payload.get("id")
f = db.get_or_404(Folder, id)
f.name = folder_name
db.session.add(f)
db.session.commit()
response = {"status": f}
return render(response)
return app
if __name__ == "__main__":
app =create_app()
app.run(host="0.0.0.0", port=5000, debug=True)