bbaovanc/InvMan
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Update all routes to use SQLAlchemy

Browse Source
This commit is contained in:
BBaoVanC 2020-10-05 01:00:50 -05:00
parent af9cdf850c
commit 8ae91f7ffa
No known key found for this signature in database
GPG Key ID: 6D74C8B0E7D791C2
4 changed files with 168 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -129,5 +129,8 @@ dmypy.json
# Pyre type checker # Pyre type checker
.pyre/ .pyre/
# VSCode custom files
.vscode/
# InvMan custom # InvMan custom
postgres.ini postgres.ini

2
.pylintrc
View File

@ -1,5 +1,5 @@
[TYPECHECK] [TYPECHECK]
ignored-modules=flask_sqlalchemy ignored-modules=flask, flask_api, sqlalchemy
generated-members=session.* generated-members=session.*
[MESSAGES CONTROL] [MESSAGES CONTROL]

22
libdb.py
View File

@ -6,7 +6,7 @@ Library for interacting with InvMan DB using SQLAlchemy
""" """
from configparser import ConfigParser from configparser import ConfigParser
from sqlalchemy import create_engine, Column, String, Text, BigInteger from sqlalchemy import create_engine, Column, String, Text, BigInteger, Float
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import sessionmaker
@ -53,3 +53,23 @@ class ProductQuantity(Base):
name = Column(Text) name = Column(Text)
quantity = Column(BigInteger) quantity = Column(BigInteger)
location = Column(String(length=32), primary_key=True) location = Column(String(length=32), primary_key=True)
class Product(Base):
"""product table"""
__tablename__ = 'product'
upc = Column(String(length=32), primary_key=True)
brand = Column(String(length=63))
name = Column(Text)
description = Column(Text)
size = Column(Float)
sizeunit = Column(String(length=32))
class Brand(Base):
"""brand table"""
__tablename__ = 'brand'
name = Column(String(length=63), primary_key=True)
description = Column(Text)

268
webapi.py
View File

@ -6,11 +6,12 @@ Web API for InvMan
""" """
# Imports # Imports
from flask import Flask, jsonify from flask import Flask, jsonify, request
from flask_api import status from flask_api import status
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
from libdb import Session, Location, ProductQuantity import sqlalchemy
from libdb import engine, Session, Location, ProductQuantity, Product, Brand
app = Flask(__name__) # app is the Flask app app = Flask(__name__) # app is the Flask app
@ -58,156 +59,173 @@ def api_get_current_quantities(location):
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
""" @app.route("/api/v1/location/<location>/quantity/<searchmethod>/<search>", methods = ["GET"]) @app.route("/api/v1/location/<location>/quantity/<searchmethod>/<search>", methods = ["GET"])
def api_get_quantity_of_product_in_location(location, searchmethod, search): def api_get_quantity_of_product_in_location(location, searchmethod, search):
db = get_db() """Route to get the quantity of a product at a location"""
with db.cursor() as cur: session = Session()
methodmap = {'upc': 'product_upc', 'name': 'name'} try:
if searchmethod in methodmap.keys(): if searchmethod == 'upc':
method = methodmap[searchmethod] data = session.query(ProductQuantity.quantity).filter(
cur.execute("SELECT quantity FROM product_quantity WHERE location = %s AND {0} = %s".format(method), (location, search)) ProductQuantity.location == location,
data = cur.fetchall() ProductQuantity.product_upc == search).one()
print("ran SELECT quantity FROM product_quantity WHERE location = %s AND {0} = %s".format(method)) return jsonify(data[0])
if len(data) <= 0: elif searchmethod == 'name':
print("no results returned, sending 404") data = session.query(ProductQuantity.quantity).filter(
return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND ProductQuantity.location == location,
else: ProductQuantity.name == search).one()
data2 = data[0][0] return jsonify(data[0])
print("processed data:")
print(data2)
return jsonify(data2)
else: else:
print("invalid search method, sending 400") print("invalid search method, sending 400")
return jsonify({'error': 'INVALID_SEARCH_METHOD'}), status.HTTP_400_BAD_REQUEST """ return jsonify({'error': 'INVALID_SEARCH_METHOD'}), status.HTTP_400_BAD_REQUEST
except NoResultFound:
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
""" @app.route("/api/v1/products", methods = ["GET"]) @app.route("/api/v1/products", methods = ["GET"])
def api_get_products(): def api_get_products():
db = get_db() """Route to get a list of products"""
with db.cursor() as cur: session = Session()
cur.execute("SELECT upc,brand,name,size,sizeunit,description FROM product") try:
data = cur.fetchall() data = {}
print("ran SELECT upc,brand,name,size,sizeunit,description FROM product") for upc, brand, name, size, sizeunit, description in session.query(
if len(data) <= 0: Product.upc, Product.brand, Product.name,
print("no results returned, sending 404") Product.size, Product.sizeunit, Product.description).all():
return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND data[upc] = {'brand': brand, 'name': name, 'size': size,
else: 'sizeunit': sizeunit, 'description': description}
data2 = {} return jsonify(data)
for row in data: except NoResultFound:
data2[row[0]] = {'brand': row[1], 'name': row[2], 'size': row[3], 'sizeunit': row[4], 'description': row[5]} return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
print("processed data:")
print(data2)
return jsonify(data2) """
""" @app.route("/api/v1/product/<searchmethod>/<search>", methods = ["GET"]) @app.route("/api/v1/product/<searchmethod>/<search>", methods = ["GET"])
def api_get_product_information(searchmethod, search): def api_get_product_information(searchmethod, search):
db = get_db() """Route to get a information about a product"""
with db.cursor() as cur: session = Session()
if searchmethod in ['name', 'upc']: try:
cur.execute("SELECT upc,brand,name,size,sizeunit,description FROM product WHERE {0} = %s".format(searchmethod), (search,)) if searchmethod == 'name':
data = cur.fetchall() result = session.query(Product.upc, Product.brand, Product.name,
print("ran SELECT upc,brand,name,size,sizeunit,description FROM product WHERE {0} = %s") Product.size, Product.sizeunit, Product.description).filter(
if len(data) <= 0: Product.name == search
print("no results returned, sending 404") ).one()
return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3],
else: 'sizeunit': result[4], 'description': result[5]}
row = data[0] return jsonify(data)
data2 = {} if searchmethod == 'upc':
data2['upc'] = row[0] result = session.query(Product.upc, Product.brand, Product.name,
data2['brand'] = row[1] Product.size, Product.sizeunit, Product.description).filter(
data2['name'] = row[2] Product.upc == search
data2['size'] = row[3] ).one()
data2['sizeunit'] = row[4] data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3],
data2['description'] = row[5] 'sizeunit': result[4], 'description': result[5]}
return jsonify(data)
print("processed data:")
print(data2)
return jsonify(data2)
else: else:
print("invalid search method, sending 400") print("invalid search method, sending 400")
return jsonify({'error': 'INVALID_SEARCH_METHOD'}), status.HTTP_400_BAD_REQUEST """ return jsonify({'error': 'INVALID_SEARCH_METHOD'}), status.HTTP_400_BAD_REQUEST
except NoResultFound:
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
""" @app.route("/api/v1/brands", methods = ["GET"]) @app.route("/api/v1/brands", methods = ["GET"])
def api_list_brands(): def api_list_brands():
db = get_db() """Route to list all brands"""
with db.cursor() as cur: session = Session()
cur.execute("SELECT name,description FROM brand") try:
data = cur.fetchall() data = {}
print("ran SELECT name,description FROM brand") for name, description in session.query(Brand.name, Brand.description).all():
if len(data) <= 0: data[name] = {'description': description}
print("no results returned, sending 404") return jsonify(data)
return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND except NoResultFound:
else: return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
data2 = {}
for row in data:
data2[row[0]] = {'description': row[1]}
print("processed data:")
print(data2)
return jsonify(data2) """
""" @app.route("/api/v1/brand/name/<search>", methods = ["GET"]) @app.route("/api/v1/brand/name/<search>", methods = ["GET"])
def api_get_brand_by_name(search): def api_get_brand_by_name(search):
db = get_db() """Route to get information about a location"""
with db.cursor() as cur: session = Session()
cur.execute("SELECT name,description FROM brand WHERE name = %s", (search,)) try:
data = cur.fetchall() data = session.query(Brand.name, Brand.description). \
print("ran SELECT name,description FROM brand WHERE name = %s") filter(Brand.name == search).one()
if len(data) <= 0: except NoResultFound:
print("no results returned, sending 404") return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND data2 = {'name': data.name, 'description': data.description}
else: return jsonify(data2)
data2 = {}
for row in data:
data2[row[0]] = {'description': row[1]}
print("processed data:")
print(data2)
return jsonify(data2) """
""" @app.route("/api/v1/create_location", methods = ["POST"]) @app.route("/api/v1/create_location", methods = ["POST"])
def api_create_location(): def api_create_location():
db = get_db() """Route to create a new location"""
with db.cursor() as cur: session = Session()
if "name" not in request.form: if "name" not in request.form:
return jsonify({'error': 'NO_NAME_PROVIDED'}), status.HTTP_400_BAD_REQUEST return jsonify({'error': 'NO_NAME_PROVIDED'}), status.HTTP_400_BAD_REQUEST
else: else:
locname = request.form['name'] locname = request.form['name']
if "description" in request.form: if "description" in request.form:
locdesc = request.form['description'] locdesc = request.form['description']
else: else:
locdesc = None locdesc = None
print(locname, locdesc) ins = sqlalchemy.insert(Location).values(name=locname, description=locdesc)
cur.execute("INSERT INTO location (name, description) VALUES (%s, %s)", (locname, locdesc)) print(f"ins is {ins}")
print("ran INSERT INTO location (name, description) VALUES (%s, %s)") session.execute(ins)
db.commit() print("executed ins")
print("committed changes to database") session.commit()
return jsonify({'api_endpoint': '/api/v1/location/name/{0}'.format(locname)}) """ print("committed")
return jsonify({'api_endpoint': f'/api/v1/location/name/{locname}'})
""" @app.route("/api/v1/create_brand", methods = ["POST"]) @app.route("/api/v1/create_brand", methods = ["POST"])
def api_create_brand(): def api_create_brand():
db = get_db() """Route to create a new brand"""
with db.cursor() as cur: session = Session()
if "name" not in request.form: if "name" not in request.form:
return jsonify({'error': 'NO_NAME_PROVIDED'}), status.HTTP_400_BAD_REQUEST return jsonify({'error': 'NO_NAME_PROVIDED'}), status.HTTP_400_BAD_REQUEST
else: else:
locname = request.form['name'] brandname = request.form['name']
if "description" in request.form: if "description" in request.form:
locdesc = request.form['description'] branddesc = request.form['description']
else: else:
locdesc = None branddesc = None
print(f"creating brand with name {locname} and description {locdesc}") ins = sqlalchemy.insert(Brand).values(name=brandname, description=branddesc)
cur.execute("INSERT INTO brand (name, description) VALUES (%s, %s)", (locname, locdesc)) print(f"ins is {ins}")
print("ran INSERT INTO brand (name, description) VALUES (%s, %s)") session.execute(ins)
db.commit() print("executed ins")
print("committed changes to database") session.commit()
return jsonify({'api_endpoint': '/api/v1/brand/name/{0}'.format(locname)}) """ print("committed")
return jsonify({'api_endpoints': [f'/api/v1/brand/name/{brandname}']})
@app.route("/api/v1/create_product", methods = ["POST"])
def api_create_product():
"""Route to create a new product"""
session = Session()
reqfields = ['upc', 'brand', 'name', 'size', 'sizeunit']
for field in reqfields:
if field not in request.form:
return jsonify({'error': f'NO_{field.upper()}_PROVIDED'}), status.HTTP_400_BAD_REQUEST
if "description" in request.form:
desc = request.form['description']
else:
desc = None
ins = sqlalchemy.insert(Product).values(upc=request.form['upc'], brand=request.form['brand'],
name=request.form['name'], size=request.form['size'],
sizeunit=request.form['sizeunit'],
description=desc)
print(f"ins is {ins}")
session.execute(ins)
print("executed ins")
session.commit()
print("committed")
return jsonify({'api_endpoints': [f'/api/v1/product/upc/{request.form["upc"]}',
f'/api/v1/product/name/{request.form["name"]}']})
if __name__ == "__main__": if __name__ == "__main__":