386 lines
9.9 KiB
Python
386 lines
9.9 KiB
Python
from flask import Flask, request, jsonify
|
|
import time
|
|
import json
|
|
import uuid
|
|
|
|
from marshmallow import Schema, fields
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
|
|
app = Flask(__name__)
|
|
app.config.from_pyfile('config.py', silent=True)
|
|
|
|
db = SQLAlchemy(app)
|
|
|
|
class UserModel(db.Model):
|
|
__tablename__ = "user"
|
|
uuid = db.Column(db.String(32), unique=True, primary_key=True, nullable=False)
|
|
name = db.Column(db.String(64), nullable=False)
|
|
bal_uuid = db.Column(db.String(32), db.ForeignKey('balance.uuid'))
|
|
|
|
class CategoryModel(db.Model):
|
|
__tablename__ = "category"
|
|
uuid = db.Column(db.String(32), unique=True, primary_key=True, nullable=False)
|
|
name = db.Column(db.String(64), nullable=False)
|
|
|
|
class RecordModel(db.Model):
|
|
__tablename__ = "record"
|
|
uuid = db.Column(db.String(32), primary_key=True, nullable=False)
|
|
user_uuid = db.Column(db.String(32), db.ForeignKey('user.uuid'))
|
|
cat_uuid = db.Column(db.String(32), db.ForeignKey('category.uuid'))
|
|
date = db.Column(db.Date)
|
|
amount = db.Column(db.Integer)
|
|
|
|
class BalanceModel(db.Model):
|
|
__tablename__ = "balance"
|
|
uuid = db.Column(db.String(32), primary_key=True, nullable=False)
|
|
value = db.Column(db.Integer, nullable=False)
|
|
|
|
|
|
class UserSchema(Schema):
|
|
uuid = fields.Str()
|
|
name = fields.Str()
|
|
bal_uuid = fields.Str()
|
|
|
|
class CategorySchema(Schema):
|
|
uuid = fields.Str()
|
|
name = fields.Str()
|
|
|
|
class RecordSchema(Schema):
|
|
uuid = fields.Str()
|
|
user_uuid = fields.Str()
|
|
cat_uuid = fields.Str()
|
|
date = fields.Date()
|
|
amount = fields.Integer()
|
|
|
|
class BalanceSchema(Schema):
|
|
uuid = fields.Str()
|
|
value = fields.Integer()
|
|
|
|
user_schema = UserSchema()
|
|
users_schema = UserSchema(many = True)
|
|
|
|
category_schema = CategorySchema()
|
|
categories_schema = CategorySchema(many = True)
|
|
|
|
record_schema = RecordSchema()
|
|
records_schema = RecordSchema(many = True)
|
|
|
|
balance_schema = BalanceSchema()
|
|
|
|
# "migration"
|
|
with app.app_context():
|
|
db.create_all()
|
|
|
|
@app.route("/healthcheck")
|
|
def ep_healthcheck():
|
|
return {
|
|
"date": time.strftime('%Y.%m.%d %H:%M:%S'),
|
|
"status": "OK"
|
|
}
|
|
|
|
@app.route("/reset_users_because_postman_is_dumb_like_that")
|
|
def ep_reset():
|
|
db.session.query(RecordModel).delete()
|
|
db.session.query(CategoryModel).delete()
|
|
db.session.query(UserModel).delete()
|
|
db.session.query(BalanceModel).delete()
|
|
db.session.commit()
|
|
return {}, 200
|
|
|
|
@app.route("/users", methods = ["GET"])
|
|
def ep_users_get():
|
|
result = db.session.query(UserModel).all()
|
|
return users_schema.dumps(result)
|
|
|
|
@app.route("/user/<user_id>", methods = ["GET"])
|
|
def ep_user_get(user_id):
|
|
result = db.session.query(UserModel).filter(UserModel.uuid == user_id).all()
|
|
|
|
if len(result) == 1:
|
|
return user_schema.dumps(result[0]), 200
|
|
else:
|
|
return {}, 404
|
|
|
|
@app.route("/user", methods = ["POST"])
|
|
def ep_user_post():
|
|
body = request.json
|
|
|
|
if not body:
|
|
return {}, 403
|
|
|
|
if 'uuid' in body:
|
|
return {}, 403
|
|
|
|
b = BalanceModel(uuid=uuid.uuid4().hex, value=0)
|
|
body.update({'uuid': uuid.uuid4().hex})
|
|
body.update({'bal_uuid': b.uuid})
|
|
|
|
try:
|
|
_ = user_schema.load(body)
|
|
except ValidationError as e:
|
|
return {}, 403
|
|
|
|
u = UserModel(**body)
|
|
|
|
try:
|
|
db.session.add(b)
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return {}, 403
|
|
|
|
return jsonify(user_schema.load(body)), 200
|
|
|
|
@app.route("/user/<user_id>", methods = ["DELETE"])
|
|
def ep_user_delete(user_id):
|
|
try:
|
|
result = db.session.query(UserModel).filter(UserModel.uuid == user_id).all()
|
|
except Exception as e:
|
|
return {}, 403
|
|
|
|
if len(result) == 0:
|
|
return {}, 404
|
|
|
|
try:
|
|
db.session.query(UserModel).filter(UserModel.uuid == user_id).delete()
|
|
db.session.query(BalanceModel).filter(BalanceModel.uuid == result[0].bal_uuid).delete()
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return {}, 403
|
|
|
|
return user_schema.dumps(result[0]), 200
|
|
|
|
@app.route("/category", methods = ["GET"])
|
|
def ep_category_get():
|
|
body = request.json
|
|
|
|
if 'uuid' in body:
|
|
result = db.session.query(CategoryModel).filter(CategoryModel.uuid == body['uuid']).all()
|
|
|
|
if len(result) == 1:
|
|
return user_schema.dumps(result[0]), 200
|
|
else:
|
|
return {}, 404
|
|
else:
|
|
return {}, 403
|
|
|
|
@app.route("/category", methods = ["POST"])
|
|
def ep_category_post():
|
|
body = request.json
|
|
|
|
if not body:
|
|
return {}, 403
|
|
|
|
if 'uuid' in body:
|
|
return {}, 403
|
|
|
|
body.update({'uuid': uuid.uuid4().hex})
|
|
|
|
try:
|
|
_ = category_schema.load(body)
|
|
except ValidationError as e:
|
|
return {}, 403
|
|
|
|
c = CategoryModel(**body)
|
|
|
|
try:
|
|
db.session.add(c)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return {}, 403
|
|
|
|
return jsonify(category_schema.load(body)), 200
|
|
|
|
@app.route("/category", methods = ["DELETE"])
|
|
def ep_category_delete():
|
|
body = request.json
|
|
|
|
if 'uuid' not in body:
|
|
return {}, 403
|
|
|
|
cat_id = body['uuid']
|
|
|
|
try:
|
|
result = db.session.query(CategoryModel).filter(CategoryModel.uuid == cat_id).all()
|
|
except Exception as e:
|
|
return {}, 403
|
|
|
|
if len(result) == 0:
|
|
return {}, 404
|
|
|
|
try:
|
|
db.session.query(CategoryModel).filter(CategoryModel.uuid == cat_id).delete()
|
|
db.session.commit()
|
|
except Exception as e:
|
|
return {}, 403
|
|
|
|
return category_schema.dumps(result[0]), 200
|
|
|
|
@app.route("/record/<record_id>", methods = ["GET"])
|
|
def ep_record_get(record_id):
|
|
result = db.session.query(RecordModel).filter(RecordModel.uuid == record_id).all()
|
|
|
|
if len(result) == 1:
|
|
return user_schema.dumps(result[0]), 200
|
|
else:
|
|
return {}, 404
|
|
|
|
@app.route("/record", methods = ["GET"])
|
|
def ep_record_get_filtered():
|
|
r = db.session.query(RecordModel)
|
|
|
|
filtered = False
|
|
|
|
if 'user_id' in request.json:
|
|
r = r.filter(RecordModel.user_uuid == request.json['user_id'])
|
|
filtered = True
|
|
elif 'user_uuid' in request.json:
|
|
r = r.filter(RecordModel.user_uuid == request.json['user_uuid'])
|
|
filtered = True
|
|
|
|
if 'cat_id' in request.json:
|
|
r = r.filter(RecordModel.cat_uuid == request.json['cat_id'])
|
|
filtered = True
|
|
if 'cat_uuid' in request.json:
|
|
r = r.filter(RecordModel.cat_uuid == request.json['cat_uuid'])
|
|
filtered = True
|
|
|
|
|
|
if filtered:
|
|
return records_schema.dumps(r.all())
|
|
else:
|
|
return [], 403
|
|
|
|
@app.route("/record/<record_id>", methods = ["DELETE"])
|
|
def ep_record_del(record_id):
|
|
try:
|
|
result = db.session.query(RecordModel).filter(RecordModel.uuid == record_id).all()
|
|
except Exception as e:
|
|
return {}, 403
|
|
|
|
if len(result) == 0:
|
|
return {}, 404
|
|
|
|
db.session.query(RecordModel).filter(RecordModel.uuid == record_id).delete()
|
|
db.session.commit()
|
|
|
|
return record_schema.dumps(result[0]), 200
|
|
|
|
@app.route("/record", methods = ["POST"])
|
|
def ep_record_post():
|
|
body = request.json
|
|
|
|
if not body:
|
|
return {}, 403
|
|
|
|
if 'uuid' in body:
|
|
return {}, 403
|
|
|
|
body.update({'uuid': uuid.uuid4().hex})
|
|
|
|
# backward compatibility with lab2 DB model
|
|
if 'cat_id' in body:
|
|
body.update({'cat_uuid': body['cat_id']})
|
|
del body['cat_id']
|
|
|
|
if 'user_id' in body:
|
|
body.update({'user_uuid': body['user_id']})
|
|
del body['user_id']
|
|
|
|
try:
|
|
_ = record_schema.load(body)
|
|
except Exception as e:
|
|
return {}, 403
|
|
|
|
r = RecordModel(**body)
|
|
|
|
b_id = db.session \
|
|
.query(UserModel) \
|
|
.filter(UserModel.uuid == body['user_uuid']) \
|
|
.all()[0] \
|
|
.bal_uuid
|
|
|
|
v = db.session \
|
|
.query(BalanceModel) \
|
|
.filter(BalanceModel.uuid == b_id) \
|
|
.all()[0] \
|
|
.value
|
|
|
|
BalanceModel.metadata.tables.get("balance").update().where(BalanceModel.metadata.tables.get("balance").c.uuid == b_id).values(value = v-body['amount'])
|
|
|
|
try:
|
|
db.session.add(r)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return {}, 403
|
|
|
|
|
|
return jsonify(record_schema.load(body)), 200
|
|
|
|
@app.route("/balance_up", methods = ["POST"])
|
|
def ep_balance_up():
|
|
body = request.json
|
|
|
|
if 'user_id' in body:
|
|
body.update({'user_uuid': body['user_id']})
|
|
del body['user_id']
|
|
elif 'uuid' in body:
|
|
body.update({'user_uuid': body['uuid']})
|
|
del body['uuid']
|
|
|
|
if 'user_uuid' not in body:
|
|
return {}, 403
|
|
|
|
try:
|
|
b_id = db.session \
|
|
.query(UserModel) \
|
|
.filter(UserModel.uuid == body['user_uuid']) \
|
|
.all()[0] \
|
|
.bal_uuid
|
|
|
|
v = db.session \
|
|
.query(BalanceModel) \
|
|
.filter(BalanceModel.uuid == b_id) \
|
|
.all()[0] \
|
|
.value
|
|
|
|
BalanceModel.metadata.tables.get("balance").update().where(BalanceModel.metadata.tables.get("balance").c.uuid == b_id).values(value = v + body['amount'])
|
|
except Exception as e:
|
|
return {}, 403
|
|
|
|
return {}, 200
|
|
|
|
@app.route("/balance", methods = ["GET"])
|
|
def ep_balance_get():
|
|
body = request.json
|
|
|
|
if 'user_id' in body:
|
|
body.update({'user_uuid': body['user_id']})
|
|
del body['user_id']
|
|
elif 'uuid' in body:
|
|
body.update({'user_uuid': body['uuid']})
|
|
del body['uuid']
|
|
|
|
if 'user_uuid' not in body:
|
|
return {}, 403
|
|
|
|
try:
|
|
b_id = db.session \
|
|
.query(UserModel) \
|
|
.filter(UserModel.uuid == body['user_uuid']) \
|
|
.all()[0] \
|
|
.bal_uuid
|
|
|
|
result = db.session.query(BalanceModel).filter(BalanceModel.uuid == b_id).all()
|
|
except Exception as e:
|
|
return {}, 403
|
|
|
|
if len(result) == 1:
|
|
return user_schema.dumps(result[0]), 200
|
|
else:
|
|
return {}, 404
|