-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathserver.py-originalexpample
104 lines (86 loc) · 2.94 KB
/
server.py-originalexpample
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
import json
import datetime
from flask import Flask, request, jsonify
from flask_mongoengine import MongoEngine
app = Flask(__name__)
app.config['MONGODB_SETTINGS'] = {
'db': 'your_database',
'host': 'ubuntu.home',
'port': 27017
}
db = MongoEngine()
db.init_app(app)
class User(db.Document):
name = db.StringField()
email = db.StringField()
def to_json(self):
return {"name": self.name,
"email": self.email}
class Sensor_data(db.Document):
sample_id = db.StringField()
time_at = db.DateTimeField()
station_id = db.StringField()
parameter = db.StringField()
time_for = db.DateTimeField()
value = db.DictField()
units = db.StringField()
def to_json(self):
return {"sample_id" : self.sample_id,
"time_at" : self.time_at,
"station_id": self.station_id,
"parameter" : self.parameter,
"time_for" : self.time_for,
"value" : self.value,
"units" : self.units}
@app.route('/', methods=['GET'])
def query_records():
sampels = Sensor_data.objects.order_by("-time_for").first()
return jsonify(sampels.to_json())
# name = request.args.get('name')
# user = User.objects(name=name).first()
# if not user:
# return jsonify({'error': 'data not found'})
# else:
# return jsonify(user.to_json())
@app.route('/', methods=['PUT'])
def create_record():
record = json.loads(request.data)
user = User(name=record['name'],
email=record['email'])
user.save()
return jsonify(user.to_json())
@app.route('/', methods=['POST'])
def update_record():
records = json.loads(request.data)
#print(records)
#print(type(records[0]))
for record in records:
print(record)
sample = Sensor_data(sample_id =record[0],
station_id=record[1],
parameter =record[2],
time_at =datetime.datetime.strptime(record[3], "%Y-%m-%dT%H:%M:%S"),
time_for=datetime.datetime.strptime(record[4], "%Y-%m-%dT%H:%M:%S"),
value =record[5],
units=record[6])
# print(record[5])
# print(type(record[5]))
sample.save()
# user = User.objects(name=record['name']).first()
# if not user:
# return jsonify({'error': 'data not found'})
# else:
# user.update(email=record['email'])
# return jsonify(user.to_json())
return jsonify({"data received" : "ok"})
@app.route('/', methods=['DELETE'])
def delete_record():
record = json.loads(request.data)
user = User.objects(name=record['name']).first()
if not user:
return jsonify({'error': 'data not found'})
else:
user.delete()
return jsonify(user.to_json())
if __name__ == "__main__":
app.run(host="0.0.0.0", debug=True)