hasslesstech
bba801f2d2
Add respecting schema, model and adapt all endpoints to utilize it. Add two new endpoints for increasing and checking the balance.
380 lines
9.7 KiB
Python
380 lines
9.7 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']
|
|
|
|
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']
|
|
|
|
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
|