The code runs fine and was manageable without any error handling.

Once I started to handle errors and build out the code the if/elif/else branches quickly became too much. How can I route the view function conditionally based on the button that sent the request without all the conditionals? I tried refactoring the get code into a helper function that could be called within the home() function but I ran into a flask 'return None' error. There has to be a better design pattern than this mess of if statements.

Am I stuck with this mess until I use a rest framework to send a function call based on the button clicked?

This is my first code review but there have to be many ways to improve this code. Please don't hold back.

from datetime import date
import json
import os
import re
import time

from bs4 import BeautifulSoup
import click
import requests
from requests.exceptions import HTTPError, Timeout
from sqlalchemy.exc import IntegrityError, SQLAlchemyError, StatementError

from project.crud import create_user, create_index, create_portfolio_balance, check_user_email
from project.database import Base, db_session, engine
from project.log import logger
from project.models import Company, DimCompanyInfo, Index, DimIndexPrice, DimCompanyPrice
from project.utils import row2dict

dashboard_blueprint = Blueprint('dashboard', __name__, url_prefix='/dashboard')

@dashboard_blueprint.route('/home', methods=('GET', 'POST'))
def home():
# dirty way of not using REST api to distinguish between buy/sell/add funds
# use button key as quasi switch case
# button 1: 'tickerBought' buys shares of a stock
# button 2: 'tickerSold' sells shares of a stock

# button 1: 'tickerBought' buys shares of a stock
if request.method == 'POST':
f = request.form
if 'tickerBought' in f.keys():
user_id = session.get('user_id')
row = {}
row['PortfolioID']= session.get('user_id')
row['Symbol']= request.form['tickerBought'].upper()
row['NumberShares'] = float(request.form['amountSharesBought'])
row['Date'] = date.fromisoformat(request.form['dateBought'])

query = db_session.query(Company, DimCompanyPrice).\
join(DimCompanyPrice, Company.ID == DimCompanyPrice.CompanyID).\
filter(DimCompanyPrice.Date <= row.get('Date')).\
filter(Company.Symbol == str.upper(row.get('Symbol'))).\
order_by(DimCompanyPrice.Date.desc()).\
first()
logger.info(query)

if query is None:
flash('Price data not available for this transaction. Pick another date.')
# check if there is enough avail balance to cover transaction
shares = row['NumberShares']
total_price = float(price*shares)
sym = row['Symbol']

query = db_session.query(PortfolioBalance).\
filter(PortfolioBalance.PortfolioID == user_id).\
first()

balance = query.Balance
if balance >= total_price:
row['TotalPrice'] = total_price
_row = Portfolio(**row)
try:
db_session.commit()
logger.info('commited')
except StatementError as e:
logger.info(f'{e} error')
try:
balance_query = db_session.query(PortfolioBalance).\
filter(PortfolioBalance.PortfolioID == user_id).\
update({'Balance': PortfolioBalance.Balance - total_price})
db_session.commit()
logger.info('balance updated')
except StatementError as e:
logger.info(f'{e} error')

# create celery task to send a transaction email
send_bought_shares_email.delay(user_id, int(shares), sym)
else:
logger.info('Not enough available balance to complete the transaction')
flash('Not enough available balance to complete the transaction')
elif 'tickerSold' in f.keys(): # button 2: 'tickerSold' sells shares of a stock
user_id = session.get('user_id')
date_sold = date.fromisoformat(request.form['dateSold'])
sym = request.form['tickerSold'].upper()
amount_shares_sold = float(request.form['amountSharesSold'])

q = db_session.query(Portfolio.Symbol, func.sum(Portfolio.NumberShares)).\
filter(Portfolio.PortfolioID == user_id).\
filter(Portfolio.Symbol == sym).\
filter(Portfolio.Date <= date_sold).\
all()

if q[0][1] is None:
flash('You do not own any shares to sell')
elif q[0][1] >= amount_shares_sold:
row = {}
row['PortfolioID']= user_id
row['Symbol']= sym
row['NumberShares'] = -amount_shares_sold
row['Date'] = date_sold

query = db_session.query(Company, DimCompanyPrice).\
join(DimCompanyPrice, Company.ID == DimCompanyPrice.CompanyID).\
filter(DimCompanyPrice.Date >= row.get('Date')).\
filter(Company.Symbol == str.upper(row.get('Symbol'))).\
order_by(DimCompanyPrice.Date.desc()).\
first()
# logger.info(query)
shares = row['NumberShares']
row['TotalPrice'] = float(price*shares)

_row = Portfolio(**row)
try:
db_session.commit()
logger.info('commited')
except StatementError as e:
logger.info(f'{e} error')
try:
balance_query = db_session.query(PortfolioBalance).\
filter(PortfolioBalance.PortfolioID == user_id).\
update({'Balance': PortfolioBalance.Balance + abs(row.get('TotalPrice'))})
db_session.commit()
logger.info('balance updated')
except StatementError as e:
logger.info(f'{e} error')

# create celery task to send a transaction email
send_sold_shares_email.delay(user_id, int(amount_shares_sold), sym)
else:
logger.info('Not enough shares owned to complete the transaction')
flash('Not enough shares owned to complete the transaction')
user_id = session.get('user_id')
else:
# check if PortfolioID exits
balance_query = db_session.query(PortfolioBalance).\
filter(PortfolioBalance.PortfolioID == user_id).\
db_session.commit()

# Get Method
row = {}
user_id = session.get('user_id')
chart_labels = []
chart_data = []

balance_query = db_session.query(PortfolioBalance).\
filter(PortfolioBalance.PortfolioID == user_id).\
first()

chart_data.append(balance_query.Balance)
chart_labels.append('Cash')

# Total price and total number of shares
payload['portfolio_cols'] = ['#','Symbol', 'Quantity', 'Total Cost', 'Current Balance', 'Return']
try:
query = db_session.query(Portfolio, func.sum(Portfolio.TotalPrice).label('TotalPrice'),\
func.sum(Portfolio.NumberShares).label('TotalShares')).\
filter(Portfolio.PortfolioID == user_id).\
group_by(Portfolio.Symbol).\
all()
# logger.info(query)
except OperationalError as e: # prevent error on empty portfolio
pass
else:
# each symbol in the portfolio
portfolio_list = []
for row in query:
_symbol = str(row[0].Symbol)
chart_labels.append(_symbol)

# current price for calculating the return
cur_price_query = db_session.query(Company, DimCompanyPrice).\
join(DimCompanyPrice, Company.ID == DimCompanyPrice.CompanyID).\
filter(Company.Symbol==_symbol).\
order_by(DimCompanyPrice.Date.desc()).\
first()

temp = {
'Symbol': _symbol,
'TotalCost': row.TotalPrice,
'Quantity': row.TotalShares,
}
balance, cost = temp.get('CurrentBalance'), temp.get('TotalCost')
_return = (balance-cost)/cost*100
temp['Return'] = _return
portfolio_list.append(temp)
chart_data.append(balance)

payload['table_cols'] = ['#','Date', 'Symbol', 'Quantity', 'Amount']

filter(Portfolio.PortfolioID == user_id).\
filter(Portfolio.NumberShares > 0).\
order_by(Portfolio.Date.desc()).\
all()

temp = {'Date': row.Date,
'Symbol': row.Symbol,
'Quantity': row.NumberShares,
'Amount': row.TotalPrice
}
# logger.info('{} {} {}'.format(row.Symbol, row.NumberShares, row.Date))

sells = db_session.query(Portfolio).\
filter(Portfolio.PortfolioID == user_id).\
filter(Portfolio.NumberShares < 0).\
order_by(Portfolio.Date.desc()).\
all()

sells_list = []
for row in sells:
temp = {'Date': row.Date,
'Symbol': row.Symbol,
'Quantity': row.NumberShares,
'Amount': row.TotalPrice
}
sells_list.append(temp)
# logger.info('{} {} {}'.format(row.Symbol, row.NumberShares, row.Date))


• Hi, welcome to Code Review. Please, do not remove the imports from your code. They are needed to understand the code Oct 10, 2021 at 0:43
• @MiguelAlorda added import code Oct 10, 2021 at 0:47

The good thing is that you are using the logger module. It is very useful for debugging too. But then you should not do this:

        except StatementError as e:
logger.info(f'{e} error')


instead use the proper logging level:

            except StatementError as e:
logger.error("Exception occurred", exc_info=True)


This will also dump the stack trace for more details (to help with troubleshooting errors).

On the other hand I am not fond of this:

except OperationalError as e:  # prevent error on empty portfolio
pass


As a rule I prefer to avoid the predictable exceptions rather than having to handle them, or worse not handling them at all. This is the type of error that could occur if your SQL is incorrect, or there is an issue with your database. So you should not be ignoring this, and you are not even keeping a trace. Find an alternative method for your test that does not involve an exception, like testing for the existence of a row.

I think that one single exception handler for the whole application would suffice, but you can still have a few try/catch blocks where you know problems may arise.

When an exception occurs, and you have started a transaction, you should rollback. It's either commit or rollback depending on the outcome. Example

I would replace query with a descriptive name for the object type you are fetching eg: ticket, portfolio etc, to make the code more explicit. Since you are using SQL Achemy you should be using field names instead of list indexes (if q[0][1] is None:).

As indicated by Reinderien, you could indeed split your route in two:

@dashboard_blueprint.route('/home', methods=('GET'))
def home():

@dashboard_blueprint.route('/home', methods=('POST'))
def home():


That will make the code more manageable. But the decorated functions should have unique names. Home is OK for the home page.

But since you present at least 3 options available to the user (buy shares, sell shares, add funds), each option should get its own route and dedicated function.

eg:

@dashboard_blueprint.route('/buy/share', methods=('POST'))

@dashboard_blueprint.route('/sell/share', methods=('POST'))
def sell_share():



To achieve this, you can also have more than one <form> in your HTML templates, to dispatch the request to the correct POST endpoint. The code for your home route is quite long because it does too many different things. Add more routes.

If you refactor your project along these lines, the benefit is that your code structure can be simplified and the two nested ifs here are no longer needed:

if request.method == 'POST':
f = request.form
if 'tickerBought' in f.keys():


Indentation is important in Python, and the more nested ifs you have, the more it is likely that you will get logic errors - especially when the code is very long, and you lose track of the correct indentation level. Those logic errors will not always be obvious. The bottom line is, keep code short and to the point, write small dedicated functions and don't abuse control structures when you can do without.

Bonus:

return render_template('dashboard/home.html', payload=payload, enumerate=enumerate)


can be written:

from flask import url_for


Instead of hardcoding a route you can refer to the associated function name, prefixed with the blueprint name. If the function names and blueprint names don't change, then you have more flexibility to rename your routes.

Warning: enumerate is the name of a built-in Python function. There is a risk that you redefine existing functions and break normal behavior. Python, like other languages, also has reserved keywords, that should not be used for identifiers (variables or function names).

Don't provide multiple methods -

methods=('GET', 'POST')


only to then check the method:

if request.method == 'POST':


Instead, separate this into two functions each accepting one method each and remove your if.

home is quite long and needs to be subdivided into multiple subroutines.

Avoid newline-escapes such as this:

            balance_query = db_session.query(PortfolioBalance).\
filter(PortfolioBalance.PortfolioID == user_id).\

balance_query = (

• How can I not provide both methods? When I go to the page I need the GET method and when I click one of the three buttons I need the POST method. That is my question exactly. How do I subdivide home() into subfunctions? Oct 10, 2021 at 17:58