-
Notifications
You must be signed in to change notification settings - Fork 0
/
Route.py
123 lines (104 loc) · 3.99 KB
/
Route.py
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
120
121
122
123
from flask import Blueprint,request,jsonify
"""
Abstract Route is about managing schema for document entries
"""
class AbstractRoute:
def __init__(self,document_name,odm):
self.document_name = document_name
self.odm = odm
self.blueprint = Blueprint(
self.document_name,
__name__,
url_prefix=f"/api/{self.document_name}"
)
self.middlewares = {}
def inject_app(self,app):
self.injected_app = app
@self.blueprint.route("/ping")
def hello_world():
return f"<p>Hello {self.document_name}!</p>"
@self.blueprint.route("/list")
def get_list():
return self.odm.list_schema()
@self.blueprint.route(f"/create",methods=["POST"])
def create():
objectId = self.odm.create_schema(request.json)
return self.odm.parse_json(request.json)
@self.blueprint.route("/read")
def read():
results = self.odm.getAll(self.document_name)
return self.odm.parse_json(results)
@self.blueprint.route("/update",methods=["PUT"])
def update():
try:
self.odm.updateOne(self.document_name,request.json['filter'],request.json["update_data"])
return {"message":[
"update succeed"
]}
except Exception as e:
return {"message":[
"update fail"
]}
# Register the blueprint in injected Flask app
app.register_blueprint(self.blueprint)
"""
Modular Route mainly for handling entries related to schema
"""
class ModularRoute:
def __init__(self,odm):
self.odm = odm
self.blueprint = Blueprint(
"modular",
__name__,
url_prefix=f"/modular"
)
def inject_app(self,app):
@self.blueprint.route(f"/api/ping")
def hello_world():
return f"<h3>Hello Modular!</h3>"
@self.blueprint.route(f"/api/count/<document_name>",methods=["GET"])
def count(document_name):
count = self.odm.count_entries(document_name)
return {"count":count,"schema":document_name}
@self.blueprint.route(f"/api/create/<document_name>",methods=["POST"])
def create(document_name):
objectId = self.odm.create(document_name,request.form.to_dict())
return request.form
@self.blueprint.route(f"/api/read/<document_name>")
def read(document_name):
results = self.odm.getByQuery(document_name,{})
return self.odm.parse_json(results)
@self.blueprint.route(f"/api/update_one/<document_name>",methods=["PUT"])
def updateOne(document_name):
try:
self.odm.updateOne(document_name,request.json['filter'],request.json["update_data"])
return {"message":[
"update succeed"
]}
except Exception as e:
print(e)
return {"message":[
"update fails"
]}
@self.blueprint.route(f"/api/delete_one/<document_name>",methods=['POST'])
def deleteOne(document_name):
try:
self.odm.deleteOne(document_name,request.json['filter'])
print("Not here")
return {"message":[
"delete succeed"
]}
except Exception as e:
print(e)
return {"message":[
"delete fails"
]}
# Register the blueprint in injected Flask app
app.register_blueprint(self.blueprint)
class SchemaRoute(AbstractRoute):
def __init__(self,odm):
self.modular_routers = {}
AbstractRoute.__init__(self,"schema_collection",odm)
def addModularRoute(self,schema):
modular_route = ModularRoute(schema,self.odm)
modular_route.inject_app(self.injected_app)