Rather embarrassingly, I have been coding for some years now, however I feel that I still write code in a very basic way. It is hard to be certain, when many companies often give coding interviews for free labour, and the internet rarely sees software development from a business side but treats it too academically fussing over details that an ordinary company would not waste resources on for productivity.
I was given a task to write in Python by a fraud prevention company, which I completed and confirmed worked correctly, within their four to five hour deadline. They gave me a list of dummy transactions within a CSV file, and the following instructions:
Using Flask + any other python libraries design a simple API service that would be able to serve information in this CSV file.
Assume this service will be part of an additional BI + ML pipeline, so propose a number of aggregates that each API could provide when queried.
This is quite an open-ended task, so feel free to take the approach you feel is right and make some assumptions if they simplify the job.
Question: How could I improve my code and solution?
Question: Without worrying about my feelings, was this good work for 4 to 5 hours? I have a university degree in Computer Science, a paid industry year where I did software development (but I learnt very little as the company used a difficult to understand framework), and have tried hard to code since I was a young child.
I assume: The answers will vary from using a framework to reduce boilerplate code, dependency injection, test cases, using several py files, more comments, use switch instead of the numerous elif statements, error handling, proper logging rather than using print for status information, and better named variables. I guess the algorithm part of the code is poor, for example it would not search the BINs very efficiently, and a RAM database might have been a faster implementation given the high traffic this could potentially receive.
Please be harsh, because after so many years I am always doing the same basic things, I want to finally code the way industries, or startup companies, expect.
Here is the source code:
#################################################################################################
# Created by XXXXX YYYYY for ZZZZZ, use at your own risk
# Test the API with Post Man by sending JSON data:
# {
# "auth": "amazingHorse",
# "action": "suseptibleBINs"
# "term": "additional information like search terms, else write 'empty'"
# }
# Setup instructions for first time users:
# Visual Studio Code
# pip3 install virtualenv
# .\env\Scripts\activate.bat
# pip3 flask flask-sqlalchemy
# pip3 install pandas
# python app.py
# !!! Warning: delete the xtransactions.sqlite file before each new deployment!!!!
#################################################################################################
from flask import Flask, request, jsonify
import sqlalchemy as db
import pandas as pd
from itertools import islice
import collections
import csv
import time
app = Flask(__name__)
adminPassword = "cyberExpert333" # Admin API password
userPassword = "amazingHorse" # User API password
suseptibleBINs = ["111","222","2a6f738d9ce1a8a9381aa775620fe5f0"]
req_data = "" # Stores the JSON received from API
# For ordinary visitors
@app.route('/')
def hello_world():
return 'Welcome, this is a private nonproduction system and provides API access to authorised users only'
# TODO admin access with special privs to drop tables etc...
@app.route('/api/admin', methods=['POST'])
def adminAccess():
req_data = request.get_json()
password = req_data['password']
if(password != adminPassword):
return "Authentication failure"
else:
return "Welcome administrator, please finish coding this..."
# Normal API access
@app.route('/api/user', methods=['POST'])
def userAccess():
req_data = request.get_json()
password = req_data['password'] # This function only checks for user password, admins have a seperate path at /api/admin
if(password != userPassword):
return "Authentication failure"
else:
action = req_data['action'] # What user would like to do
term = req_data['term'] # Optional information, such as text to search for
if(action == "BIN"):
return binsFraudRisk()
elif (action == "search"):
return searchDatabase(term)
elif (action == "mean"):
return meanMaths(term)
elif (action == "mode"):
return modeMaths(term)
elif (action == "median"):
return medianMaths(term)
elif (action == "max"):
return maxMaths(term)
elif (action == "min"):
return minMaths(term)
else:
return "Error in your JSON data, please resend request"
# Boiler plate database code
def boilerPlateDatabase():
engine = db.create_engine('sqlite:///xtransactions.sqlite')
connection = engine.connect()
metadata = db.MetaData()
census = db.Table('trans', metadata, autoload=True, autoload_with=engine)
query = db.select([census])
ResultProxy = connection.execute(query)
ResultSet = ResultProxy.fetchall()
return ResultSet
# BINs suseptible to fraud
def binsFraudRisk():
ResultSet = boilerPlateDatabase()
foundRow = list()
for row in ResultSet:
for badBIN in suseptibleBINs:
if badBIN == row[4]:
foundRow.append(row) # Add into foundRow array
return str(foundRow) # All discovered rows which have suseptible BINs
# Search string - case sensitive!
def searchDatabase(searchTerm):
ResultSet = boilerPlateDatabase()
foundRow = list()
for row in ResultSet:
if (searchTerm == row[1] or searchTerm == row[2] or searchTerm == row[3] or searchTerm == row[4] or searchTerm == row[5] or searchTerm == row[6] or searchTerm == row[7] or searchTerm == row[8] or searchTerm == row[9]):
foundRow.append(row)
return str(foundRow) # All rows which match the query
# Mean is the average
def meanMaths(searchTerm):
ResultSet = boilerPlateDatabase()
searchNumber = 0
# 7
if(searchTerm == "amount"):
searchNumber = 7
# 8
elif(searchTerm == "currency"):
searchNumber = 8
else:
return "Error in your JSON"
# New array with just with numbers from amount or currency
num = []
for number in ResultSet:
num.append(number[searchNumber])
# Calculate the mean
sum_num = 0
for t in num:
sum_num = sum_num + t
avg = sum_num / len(num)
return str(avg)
# Mode is number that appears the most
def modeMaths(searchTerm):
ResultSet = boilerPlateDatabase()
searchNumber = 0
# 7
if(searchTerm == "amount"):
searchNumber = 7
# 8
elif(searchTerm == "currency"):
searchNumber = 8
else:
return "Error in your JSON"
# New array with just with numbers from amount or currency
num_list = []
for number in ResultSet:
num_list.append(number[searchNumber])
# calculate the frequency of each item
data = collections.Counter(num_list)
data_list = dict(data)
max_value = max(list(data.values()))
mode_val = [num for num, freq in data_list.items() if freq == max_value]
if len(mode_val) == len(num_list):
return "Mode not found!"
else:
return("The Mode of the list is : " + ', '.join(map(str, mode_val)))
# Median is middle value
def medianMaths(searchTerm):
ResultSet = boilerPlateDatabase()
searchNumber = 0
# 7
if(searchTerm == "amount"):
searchNumber = 7
# 8
elif(searchTerm == "currency"):
searchNumber = 8
else:
return "Error in your JSON"
# New array with just with numbers from amount or currency
num_list = []
for number in ResultSet:
num_list.append(number[searchNumber])
# Sort the list
num_list.sort()
# Finding the position of the median
if len(num_list) % 2 == 0:
first_median = num_list[len(num_list) // 2]
second_median = num_list[len(num_list) // 2 - 1]
median = (first_median + second_median) / 2
else:
median = num_list[len(num_list) // 2]
return("The median is: " + str(median))
# Max is highest number
def maxMaths(searchTerm):
ResultSet = boilerPlateDatabase()
searchNumber = 0
# 7
if(searchTerm == "amount"):
searchNumber = 7
# 8
elif(searchTerm == "currency"):
searchNumber = 8
else:
return "Error in your JSON"
# New array with just with numbers from amount or currency
num_list = []
for number in ResultSet:
num_list.append(number[searchNumber])
return 'Maximum: '+str(max(num_list))
# Min is smallest number
def minMaths(searchTerm):
ResultSet = boilerPlateDatabase()
searchNumber = 0
# 7
if(searchTerm == "amount"):
searchNumber = 7
# 8
elif(searchTerm == "currency"):
searchNumber = 8
else:
return "Error in your JSON"
# New array with just with numbers from amount or currency
num_list = []
for number in ResultSet:
num_list.append(number[searchNumber])
return 'Minimum: '+str(min(num_list))
def databaseManagement():
print("Creating new database")
engine = db.create_engine('sqlite:///xtransactions.sqlite') # Create test.sqlite automatically
connection = engine.connect()
metadata = db.MetaData()
trans = db.Table('trans', metadata,
db.Column('Id', db.Integer(), unique=True), # Not in CSV file but makes life easier
db.Column('t', db.String(), nullable=False),
db.Column('tx_id', db.String(), nullable=False),
db.Column('src_card', db.String(), nullable=False),
db.Column('src_BIN', db.String(), nullable=False),
db.Column('dst_card', db.String(), nullable=False),
db.Column('dst_BIN', db.String(), nullable=False),
db.Column('amount', db.Integer(), nullable=False),
db.Column('currency', db.Integer(), nullable=False),
db.Column('status', db.String(), nullable=False),
)
metadata.create_all(engine) # Creates the table
# Inserting record one by one with a FOR loop
print("Loading CSV into database")
theIDCounter = 0 # Counter variable to assign a unique ID for DB
with open('trx_sample.csv') as fd:
for row in islice(csv.reader(fd), 2, None): # Skip line 0 (the disclaimer) and 1 (which is stuff like t, tx_id)
query = db.insert(trans).values(Id=theIDCounter, t=row[0], tx_id=row[1], src_card=row[2], src_BIN=row[3], dst_card=row[4], dst_BIN=row[5], amount=row[6], currency=row[7], status=row[8])
ResultProxy = connection.execute(query)
theIDCounter = theIDCounter + 1
results = connection.execute(db.select([trans])).fetchall()
if __name__ == '__main__':
databaseManagement() # Loads CSV and puts everything into the DB
app.run(debug=False) # If true then it causes server to restart which upsets DB