Add initial server implementation

This commit is contained in:
Danila Fedorin 2020-02-01 14:29:56 -08:00
commit e0136a3517
4 changed files with 145 additions and 0 deletions

13
fgs/__init__.py Normal file
View File

@ -0,0 +1,13 @@
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///./data.sqlite'
app.secret_key = os.environ.get('FGS_SECRET_KEY') or 'default secret key'
db = SQLAlchemy(app)
from .views import *
from .model import *
from .jwt import *

32
fgs/jwt.py Normal file
View File

@ -0,0 +1,32 @@
import jwt
from flask import request, g, abort
from . import db, app
from .model import User
from functools import wraps
def get_jwt(self):
return jwt.encode({'id': self.id}, app.secret_key, algorithm='HS256')
User.get_jwt = get_jwt
def jwt_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if 'Authorization' not in request.headers:
abort(400)
authorization = request.headers['Authorization'].split(' ')
if len(authorization) < 2:
abort(400)
try:
decoded = jwt.decode(
authorization[1].encode(),
app.secret_key,
algorithms = ['HS256'])
g.user = User.query.filter_by(id=decoded['id']).first()
if g.user is None: abort(400)
return f(*args, **kwargs)
except Exception as e:
print(e)
abort(500)
return decorated_function

48
fgs/model.py Normal file
View File

@ -0,0 +1,48 @@
from . import db
import bcrypt
class Collar(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String)
active = db.Column(db.Boolean)
data_points = db.relationship('DataPoint')
class DataPoint(db.Model):
id = db.Column(db.Integer, primary_key=True)
collar_id = db.Column(db.Integer, db.ForeignKey('collar.id'))
longitude = db.Column(db.Float(precision=10))
latitude = db.Column(db.Float(precision=10))
battery_level = db.Column(db.Integer)
datetime = db.Column(db.DateTime)
outside = db.Column(db.Boolean)
def to_dict(self):
return {'longitude': self.longitude,
'latitude': self.latitude,
'battery_level': self.battery_level,
'datetime': self.datetime,
'outside': self.outside
}
class StimulusActivation(db.Model):
id = db.Column(db.Integer, primary_key=True)
longitude = db.Column(db.Float(precision=10))
latitude = db.Column(db.Float(precision=10))
volume_level = db.Column(db.Integer)
voltage_level = db.Column(db.Integer)
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String)
password_hash = db.Column(db.String)
def get_password(self):
return self.password_hash
def set_password(self, password):
self.password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
def check_password(self, password):
return bcrypt.checkpw(password.encode(), self.password_hash)
password = property(get_password, set_password)

52
fgs/views.py Normal file
View File

@ -0,0 +1,52 @@
from . import app
from .jwt import jwt_required
from .model import *
from flask import g, jsonify, request, abort
from sqlalchemy import func
@app.route('/')
def index():
return 'Hello, world!'
@app.route('/login', methods=['POST'])
def login():
username = request.form.get('username')
password = request.form.get('password')
if username is None or password is None: abort(400)
user = User.query.filter_by(username=username).first()
if user is None or not user.check_password(password): abort(400)
return jsonify({ 'token': user.get_jwt().decode() })
@app.route('/me')
@jwt_required
def me():
return jsonify({ 'username' : g.user.username })
@app.route('/collars')
@jwt_required
def collars():
active_collars = Collar.query.\
filter_by(active=True).\
join(DataPoint, isouter=True).\
order_by(DataPoint.datetime.desc()).\
group_by(Collar.id).\
with_entities(
Collar.id, Collar.name,
DataPoint.longitude, DataPoint.latitude).\
all()
active_collars = [
{'id': id, 'name': name, 'pos': {'longitude': lo, 'latitude': la}}
for (id, name, lo, la) in active_collars]
return jsonify(active_collars)
@app.route('/collars/<int:id>/history')
@jwt_required
def collar_history(id):
collar = Collar.query.filter_by(id=id).first()
if collar is None: abort(404)
data_points = DataPoint.query.\
filter_by(collar_id=id).\
order_by(DataPoint.datetime.desc()).\
all()
return jsonify([point.to_dict() for point in data_points])