115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
__author__ = "Florian Kaiser"
|
|
__copyright__ = "Copyright 2022, Project Aktienbot"
|
|
__credits__ = ["Florian Kaiser", "Florian Kellermann", "Linus Eickhof", "Kevin Pauer"]
|
|
__license__ = "GPL 3.0"
|
|
__version__ = "1.0.0"
|
|
|
|
import datetime
|
|
import os
|
|
|
|
from apiflask import abort, APIBlueprint
|
|
|
|
from app.auth import auth
|
|
from app.db import database as db
|
|
from app.helper_functions import make_response, get_email_or_abort_401
|
|
from app.models import Transaction
|
|
from app.schema import TransactionSchema, TransactionResponseSchema
|
|
|
|
transaction_blueprint = APIBlueprint('transaction', __name__, url_prefix='/api')
|
|
__location__ = os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__)))
|
|
|
|
|
|
@transaction_blueprint.route('/transaction', methods=['POST'])
|
|
@transaction_blueprint.output(TransactionResponseSchema(), 200)
|
|
@transaction_blueprint.input(schema=TransactionSchema)
|
|
@transaction_blueprint.auth_required(auth)
|
|
@transaction_blueprint.doc(summary="Adds new transaction", description="Adds new transaction for current user")
|
|
def add_transaction(data):
|
|
email = get_email_or_abort_401()
|
|
|
|
# Check if required data is available
|
|
if not check_if_symbol_data_exists(data):
|
|
abort(400, "Symbol missing")
|
|
|
|
if not check_if_time_data_exists(data):
|
|
abort(400, "Time missing")
|
|
|
|
if not check_if_count_data_exists(data):
|
|
abort(400, "Count missing")
|
|
|
|
if not check_if_price_data_exists(data):
|
|
abort(400, "Price missing")
|
|
|
|
# Add transaction
|
|
new_transaction = Transaction(
|
|
email=email,
|
|
symbol=data['symbol'],
|
|
time=datetime.datetime.strptime(data['time'], '%Y-%m-%dT%H:%M:%S.%fZ'),
|
|
count=data['count'],
|
|
price=data['price']
|
|
)
|
|
db.session.add(new_transaction)
|
|
db.session.commit()
|
|
|
|
return make_response(new_transaction.as_dict(), 200, "Successfully added transaction")
|
|
|
|
|
|
@transaction_blueprint.route('/transactions', methods=['GET'])
|
|
@transaction_blueprint.output(TransactionSchema(), 200)
|
|
@transaction_blueprint.auth_required(auth)
|
|
@transaction_blueprint.doc(summary="Returns all transactions", description="Returns all transactions for current user")
|
|
def get_transaction():
|
|
email = get_email_or_abort_401()
|
|
|
|
return_transactions = []
|
|
|
|
# Get all transactions
|
|
transactions = db.session.query(Transaction).filter_by(email=email).all()
|
|
|
|
# Iterate over transactions and add them to return_transactions
|
|
if transactions is not None:
|
|
for row in transactions:
|
|
return_transactions.append(row.as_dict())
|
|
|
|
return make_response(return_transactions, 200, "Successfully loaded transactions")
|
|
|
|
|
|
def check_if_symbol_data_exists(data):
|
|
if "symbol" not in data:
|
|
return False
|
|
|
|
if data['symbol'] == "" or data['symbol'] is None:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def check_if_time_data_exists(data):
|
|
if "time" not in data:
|
|
return False
|
|
|
|
if data['time'] == "" or data['time'] is None:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def check_if_count_data_exists(data):
|
|
if "count" not in data:
|
|
return False
|
|
|
|
if data['count'] == "" or data['count'] is None:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def check_if_price_data_exists(data):
|
|
if "price" not in data:
|
|
return False
|
|
|
|
if data['price'] == "" or data['price'] is None:
|
|
return False
|
|
|
|
return True
|