Flask static url view

Question

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.

Edit: added import lines

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

from bs4 import BeautifulSoup
import click
from flask import Blueprint, current_app
import requests
from requests.exceptions import HTTPError, Timeout
from sqlalchemy.exc import IntegrityError, SQLAlchemyError, StatementError
from werkzeug.security import check_password_hash, generate_password_hash


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'))
@login_required
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 3: 'addFunds' add funds to your portfolio

    # 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
            price = query[1].CloseAdjusted
            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.add(_row)
                    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)
                price = query[1].CloseAdjusted
                shares = row['NumberShares']
                row['TotalPrice'] = float(price*shares)

                _row = Portfolio(**row)
                try:
                    db_session.add(_row)
                    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')
        else: # button 3: 'addFunds' add funds to your portfolio
            user_id = session.get('user_id')
            addFunds = float(request.form['addFunds'])
            if addFunds < 0:
                flash('Cannot add negative funds')
            else:    
                # check if PortfolioID exits
                balance_query = db_session.query(PortfolioBalance).\
                    filter(PortfolioBalance.PortfolioID == user_id).\
                    update({'Balance': PortfolioBalance.Balance + addFunds})
                db_session.commit()

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

    balance_query = db_session.query(PortfolioBalance).\
                filter(PortfolioBalance.PortfolioID == user_id).\
                first()
    payload['balance'] = balance_query.Balance

    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,
                'CurrentBalance': row.TotalShares*cur_price_query[1].CloseAdjusted,
                'CurrentPrice': cur_price_query[1].CloseAdjusted,
            }
            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['portfolio'] = portfolio_list
        payload['chart_labels'] = chart_labels
        payload['chart_data'] = chart_data

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

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

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

        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))
        payload['sells'] = sells_list

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

Answer 1

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'))
def buy_share():

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

@dashboard_blueprint.route('/add/funds', methods=('POST'))
def add_funds():

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
return render_template(url_for("dashboard_blueprint.home"), payload=payload, enumerate=enumerate)

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).

Answer 2

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).\
                update({'Balance': PortfolioBalance.Balance + addFunds})

and prefer parens:

balance_query = (
    db_session.query(PortfolioBalance)
    .filter(PortfolioBalance.PortfolioID == user_id)
    .update({'Balance': PortfolioBalance.Balance + addFunds}
)