tobias/lrc-backend
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

addded example models

Browse Source
This commit is contained in:
Tobias
2019-03-12 09:17:55 +01:00
parent 38ab1578b0
commit 3c4e3719d7
6 changed files with 377 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Pipfile
View File

@@ -8,6 +8,9 @@ flask = "*"
flask-httpauth = "*"
flask-restplus-patched = "*"
flask-sqlalchemy = "*"
flask-login = "*"
pyjwt = "*"
passlib = "*"
[dev-packages]

25
Pipfile.lock generated
View File

@@ -1,7 +1,7 @@
{
"_meta": {
"hash": {
"sha256": "b3e3f1a3dbe6801a4a969a2b5c9b73167ea078c1916db723fdc282bc9af814a7"
"sha256": "196d1a010314c8f16ccc747ed35b4821e8e3aa63f90ec2893123ea19c95935b1"
},
"pipfile-spec": 6,
"requires": {
@@ -60,6 +60,13 @@
"index": "pypi",
"version": "==3.2.4"
},
"flask-login": {
"hashes": [
"sha256:c815c1ac7b3e35e2081685e389a665f2c74d7e077cb93cecabaea352da4752ec"
],
"index": "pypi",
"version": "==0.4.1"
},
"flask-marshmallow": {
"hashes": [
"sha256:75c9d80f22af982b1e8ccec109d3b75c14bb5570602ae3705a4ff775badd2816",
@@ -150,6 +157,22 @@
],
"version": "==2.18.1"
},
"passlib": {
"hashes": [
"sha256:3d948f64138c25633613f303bcc471126eae67c04d5e3f6b7b8ce6242f8653e0",
"sha256:43526aea08fa32c6b6dbbbe9963c4c767285b78147b7437597f992812f69d280"
],
"index": "pypi",
"version": "==1.7.1"
},
"pyjwt": {
"hashes": [
"sha256:5c6eca3c2940464d106b99ba83b00c6add741c9becaec087fb7ccdefea71350e",
"sha256:8d59a976fb773f3e6a39c85636357c4f0e242707394cadadd9814f5cbaa20e96"
],
"index": "pypi",
"version": "==1.7.1"
},
"pyrsistent": {
"hashes": [
"sha256:3ca82748918eb65e2d89f222b702277099aca77e34843c5eb9d52451173970e2"

7
__init__.py
View File

@@ -1,5 +1,8 @@
# -*- coding: utf-8 -*-
import os
"""
Backend base module
"""
from flask import Flask
from flask_httpauth import HTTPTokenAuth, HTTPBasicAuth
from flask_sqlalchemy import SQLAlchemy
@@ -7,7 +10,7 @@ from flask_sqlalchemy import SQLAlchemy
from .serve_frontend import fe_bp
from .api import api_bp
jwt_auth = HTTPTokenAuth('Bearer')
jwt_auth = HTTPTokenAuth()
basic_auth = HTTPBasicAuth()
app = Flask(__name__)

11
models/example_model.py
View File

@@ -1,10 +1,15 @@
# -*- coding: utf-8 -*-
from backend import db
import uuid
class Sensor(db.Model):
class ExampleDataItem(db.Model):
"""
just an example class...
"""
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
mac = db.Column(db.String(32), nullable=False, unique=True, index=True)
uuid = db.Column(db.String(36), nullable=False, unique=True, index=True, default=str(uuid.uuid4()))
tags = db.relationship("SensorTag", secondary=sensor_tag_relations, back_populates="sensors")
rough_geo_location = db.Column(db.String, nullable=True, index=True)
some_string_value = db.Column(db.String, nullable=True, index=True)
name = db.Column(db.String(128), default="<not set>", nullable=False, index=True, unique=False)
description = db.Column(db.String(4096), nullable=True, unique=False)

19
models/post_model.py Normal file
View File

@@ -0,0 +1,19 @@
# -*- coding: utf-8 -*-
"""
Example post model and related models
"""
from backend import db
class Post(db.Model):
"""
A post example class
"""
id = db.Column(db.Integer, primary_key=True)
body = db.Column(db.String(140))
timestamp = db.Column(db.DateTime)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
def __repr__(self):
return '<Post %r>' % self.body

318
models/user_model.py Normal file
View File

@@ -0,0 +1,318 @@
# -*- coding: utf-8 -*-
"""
Example user model and related models
"""
from backend import db, app
from backend.models.post_model import Post
from backend.models.example_model import ExampleDataItem
import re
import jwt
from flask_login import UserMixin
from sqlalchemy import or_
from datetime import datetime, timedelta
from passlib.hash import sha256_crypt
from hashlib import md5
followers = db.Table('followers',
db.Column('follower_id', db.Integer, db.ForeignKey('user.id')),
db.Column('followed_id', db.Integer, db.ForeignKey('user.id'))
)
acquaintances = db.Table('acquaintances',
db.Column('me_id', db.Integer, db.ForeignKey('user.id')),
db.Column('acquaintance_id', db.Integer, db.ForeignKey('user.id'))
)
class User(UserMixin, db.Model):
"""
Example user model representation.
"""
id = db.Column(db.Integer, primary_key=True)
social_id = db.Column(db.String(64), nullable=True, unique=True)
nickname = db.Column(db.String(64), index=True, unique=True)
first_name = db.Column(db.String(64), index=True, nullable=True)
last_name = db.Column(db.String(64), index=True, nullable=True)
email = db.Column(db.String(120), nullable=False, index=True, unique=True)
lang = db.Column(db.String(16), index=False, unique=False)
timezone = db.Column(db.String(32), index=False, unique=False)
posts = db.relationship('Post', backref='author', lazy='dynamic')
example_data_item = db.relationship('ExampleDataItem', backref='owner', lazy='dynamic')
about_me = db.Column(db.String(140))
role = db.Column(db.String(64))
password = db.Column(db.String(255), nullable=True)
registered_on = db.Column(db.DateTime, nullable=False, default=datetime.utcnow())
last_seen = db.Column(db.DateTime, default=datetime.utcnow())
jwt_exp_delta_seconds = db.Column(db.Integer, nullable=True)
acquainted = db.relationship('User',
secondary=acquaintances,
primaryjoin=(acquaintances.c.me_id == id),
secondaryjoin=(acquaintances.c.acquaintance_id == id),
backref=db.backref('acquaintances', lazy='dynamic'),
lazy='dynamic')
followed = db.relationship('User',
secondary=followers,
primaryjoin=(followers.c.follower_id == id),
secondaryjoin=(followers.c.followed_id == id),
backref=db.backref('followers', lazy='dynamic'),
lazy='dynamic')
@staticmethod
def get_by_identifier(identifier):
"""
Find user by identifier, which might be the nickname or the e-mail address.
:param identifier:
:return:
"""
return User.query.filter(or_(User.nickname == identifier,
User.email == identifier)).first()
@staticmethod
def get_all():
"""
Return all users
:return:
"""
return User.query.all()
@staticmethod
def make_unique_nickname(nickname):
"""
Add suffix (counter) to nickname in order to get a unique nickname.
:param nickname:
:return:
"""
if User.query.filter_by(nickname=nickname).first() is None:
return nickname
version = 2
while True:
new_nickname = nickname + str(version)
if User.query.filter_by(nickname=new_nickname).first() is None:
break
version += 1
return new_nickname
@staticmethod
def make_valid_nickname(nickname):
"""
Replaces certain characters (except a-zA-Z0-9_.) in nickname with blancs.
:param nickname:
:return:
"""
return re.sub('[^a-zA-Z0-9_.]', '', nickname)
@property
def is_authenticated(self):
"""
Returns true if user is authenticated.
:return:
"""
# TODO: implement correctly
return True
@property
def is_active(self):
"""
Returns true if user is active.
:return:
"""
# TODO: implement correctly
return True
@property
def is_anonymous(self):
"""
Returns true if user is anonymous.
:return:
"""
# TODO: implement correctly
return False
@staticmethod
def decode_auth_token(auth_token):
"""
Decodes the auth token
:param auth_token:
:return: integer|string
"""
try:
payload = jwt.decode(auth_token, app.config.get('SECRET_KEY'))
is_blacklisted_token = BlacklistToken.check_blacklist(auth_token)
if is_blacklisted_token:
return 'Token blacklisted. Please log in again.'
else:
return payload['sub']
except jwt.ExpiredSignatureError:
return 'Signature expired. Please log in again.'
except jwt.InvalidTokenError:
return 'Invalid token. Please log in again.'
def encode_auth_token(self):
"""
Generates the Auth Token
:return: string
"""
try:
payload = {
'exp': datetime.utcnow() + timedelta(days=0, hours=3, seconds=5),
'iat': datetime.utcnow(),
'sub': self.id
}
return jwt.encode(
payload,
app.config.get('SECRET_KEY'),
algorithm='HS256'
)
except Exception as e:
return e
def set_password(self, password):
"""
SHA256 encrypts the given password and sets it on the user.
:param password:
:return:
"""
self.password = sha256_crypt.encrypt(password)
def verify_password(self, password):
"""
Verifies that the given password matches the SHA256 encrypted password stored on the user.
:param password:
:return:
"""
return sha256_crypt.verify(password, self.password)
def get_id(self):
"""
Returns the ID of the user.
:return:
"""
try:
# noinspection PyUnresolvedReferences
return unicode(self.id) # python 2
except NameError:
return str(self.id) # python 3
def avatar(self, size):
"""
Returns an avatar URL.
:param size:
:return:
"""
return 'https://s.gravatar.com/avatar/%s?d=mm&s=%d' % (md5(self.email.encode('utf-8')).hexdigest(), size)
def acquaint(self, user):
"""
Adds an acquaintance to the user object.
:param user:
:return:
"""
if not self.is_acquainted(user):
self.acquainted.append(user)
return self
def unacquaint(self, user):
"""
Removes the user from the list of acquaintances.
:param user:
:return:
"""
if self.is_acquainted(user):
self.acquainted.remove(user)
return self
def is_acquainted(self, user):
"""
Check if the provided user is an acquaintance.
:param user:
:return:
"""
return self.acquainted.filter(acquaintances.c.acquaintance_id == user.id).count() > 0
def get_acquaintances(self):
"""
Returns the list of acquaintances.
:return:
"""
return User.query.join(acquaintances, (acquaintances.c.acquaintance_id == User.id)).filter(
acquaintances.c.me_id == self.id).order_by(User.nickname.desc())
def shared_example_data_items(self):
"""
Returns a list of the shared data items.
:return:
"""
return ExampleDataItem.query.join(acquaintances,
(acquaintances.c.acquaintance_id == ExampleDataItem.user_id)).filter(
acquaintances.c.me_id == self.id).order_by(ExampleDataItem.timestamp.desc())
def follow(self, user):
"""
Add user to list of followers.
:param user:
:return:
"""
if not self.is_following(user):
self.followed.append(user)
return self
def unfollow(self, user):
"""
Remove user from the list of followers.
:param user:
:return:
"""
if self.is_following(user):
self.followed.remove(user)
return self
def is_following(self, user):
"""
Checks if specified user is a follower.
:param user:
:return:
"""
return self.followed.filter(followers.c.followed_id == user.id).count() > 0
def followed_posts(self):
"""
Returns list of followed posts.
:return:
"""
return Post.query.join(followers, (followers.c.followed_id == Post.user_id)).filter(
followers.c.follower_id == self.id).order_by(Post.timestamp.desc())
def __repr__(self):
return '<User %r>' % self.nickname
class BlacklistToken(db.Model):
"""
Token Model for storing JWT tokens
"""
__tablename__ = 'blacklist_tokens'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
token = db.Column(db.String(500), unique=True, nullable=False)
blacklisted_on = db.Column(db.DateTime, nullable=False)
def __init__(self, token):
self.token = token
self.blacklisted_on = datetime.now()
def __repr__(self):
return '<id: token: {}'.format(self.token)
@staticmethod
def check_blacklist(auth_token):
"""
check whether auth token has been blacklisted
:param auth_token:
:return:
"""
res = BlacklistToken.query.filter_by(token=str(auth_token)).first()
if res:
return True
else:
return False