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

Reformat to follow PEP standards

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

5
libdb.py
View File

@ -24,14 +24,15 @@ def load_config():
for param in params: for param in params:
config[param[0]] = param[1] config[param[0]] = param[1]
else: else:
raise Exception('Section {0} not found in the {1} file'.format(section, filename)) raise Exception(
'Section {0} not found in the {1} file'.format(section, filename))
return config return config
cfg = load_config() cfg = load_config()
engine = create_engine(f"postgresql://{cfg['user']}:{cfg['pass']}@{cfg['host']}/{cfg['database']}", engine = create_engine(f"postgresql://{cfg['user']}:{cfg['pass']}@{cfg['host']}/{cfg['database']}",
echo='debug') echo='debug')
Base = declarative_base() Base = declarative_base()
Session = sessionmaker() Session = sessionmaker()
Session.configure(bind=engine) Session.configure(bind=engine)

67
webapi.py
View File

@ -11,32 +11,32 @@ from flask_api import status
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
import sqlalchemy import sqlalchemy
from libdb import engine, Session, Location, ProductQuantity, Product, Brand from libdb import Session, Location, ProductQuantity, Product, Brand
app = Flask(__name__) # app is the Flask app app = Flask(__name__) # app is the Flask app
@app.route("/api/v1/locations", methods = ["GET"]) @app.route("/api/v1/locations", methods=["GET"])
def get_locations(): def get_locations():
"""Route to get locations""" """Route to get locations"""
session = Session() session = Session()
data = {} data = {}
try: try:
for name,desc in session.query(Location.name, Location.description): for name, desc in session.query(Location.name, Location.description):
data[name] = {'description': desc} data[name] = {'description': desc}
except NoResultFound: except NoResultFound:
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
return jsonify(data) return jsonify(data)
@app.route("/api/v1/location/name/<search>", methods = ["GET"]) @app.route("/api/v1/location/name/<search>", methods=["GET"])
def api_get_location_information(search): def api_get_location_information(search):
"""Route to get information about a location""" """Route to get information about a location"""
session = Session() session = Session()
try: try:
data = session.query(Location.name, Location.description). \ data = session.query(Location.name, Location.description). \
filter(Location.name == search).one() filter(Location.name == search).one()
except NoResultFound: except NoResultFound:
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
data2 = {} data2 = {}
@ -45,34 +45,34 @@ def api_get_location_information(search):
return jsonify(data2) return jsonify(data2)
@app.route("/api/v1/location/<location>/quantities", methods = ["GET"]) @app.route("/api/v1/location/<location>/quantities", methods=["GET"])
def api_get_current_quantities(location): def api_get_current_quantities(location):
"""Route to get quantities in a location""" """Route to get quantities in a location"""
session = Session() session = Session()
data = {} data = {}
try: try:
for upc,quantity in session.query(ProductQuantity.product_upc, ProductQuantity.quantity) \ for upc, quantity in session.query(ProductQuantity.product_upc, ProductQuantity.quantity) \
.filter(ProductQuantity.location == location).all(): .filter(ProductQuantity.location == location).all():
data[upc] = quantity data[upc] = quantity
return jsonify(data) return jsonify(data)
except NoResultFound: except NoResultFound:
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):
"""Route to get the quantity of a product at a location""" """Route to get the quantity of a product at a location"""
session = Session() session = Session()
try: try:
if searchmethod == 'upc': if searchmethod == 'upc':
data = session.query(ProductQuantity.quantity).filter( data = session.query(ProductQuantity.quantity).filter(
ProductQuantity.location == location, ProductQuantity.location == location,
ProductQuantity.product_upc == search).one() ProductQuantity.product_upc == search).one()
return jsonify(data[0]) return jsonify(data[0])
elif searchmethod == 'name': elif searchmethod == 'name':
data = session.query(ProductQuantity.quantity).filter( data = session.query(ProductQuantity.quantity).filter(
ProductQuantity.location == location, ProductQuantity.location == location,
ProductQuantity.name == search).one() ProductQuantity.name == search).one()
return jsonify(data[0]) return jsonify(data[0])
else: else:
print("invalid search method, sending 400") print("invalid search method, sending 400")
@ -82,15 +82,15 @@ def api_get_quantity_of_product_in_location(location, searchmethod, search):
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/products", methods = ["GET"]) @app.route("/api/v1/products", methods=["GET"])
def api_get_products(): def api_get_products():
"""Route to get a list of products""" """Route to get a list of products"""
session = Session() session = Session()
try: try:
data = {} data = {}
for upc, brand, name, size, sizeunit, description in session.query( for upc, brand, name, size, sizeunit, description in session.query(
Product.upc, Product.brand, Product.name, Product.upc, Product.brand, Product.name,
Product.size, Product.sizeunit, Product.description).all(): Product.size, Product.sizeunit, Product.description).all():
data[upc] = {'brand': brand, 'name': name, 'size': size, data[upc] = {'brand': brand, 'name': name, 'size': size,
'sizeunit': sizeunit, 'description': description} 'sizeunit': sizeunit, 'description': description}
return jsonify(data) return jsonify(data)
@ -98,25 +98,25 @@ def api_get_products():
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/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):
"""Route to get a information about a product""" """Route to get a information about a product"""
session = Session() session = Session()
try: try:
if searchmethod == 'name': if searchmethod == 'name':
result = session.query(Product.upc, Product.brand, Product.name, result = session.query(Product.upc, Product.brand, Product.name,
Product.size, Product.sizeunit, Product.description).filter( Product.size, Product.sizeunit, Product.description).filter(
Product.name == search Product.name == search
).one() ).one()
data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3], data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3],
'sizeunit': result[4], 'description': result[5]} 'sizeunit': result[4], 'description': result[5]}
return jsonify(data) return jsonify(data)
if searchmethod == 'upc': if searchmethod == 'upc':
result = session.query(Product.upc, Product.brand, Product.name, result = session.query(Product.upc, Product.brand, Product.name,
Product.size, Product.sizeunit, Product.description).filter( Product.size, Product.sizeunit, Product.description).filter(
Product.upc == search Product.upc == search
).one() ).one()
data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3], data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3],
'sizeunit': result[4], 'description': result[5]} 'sizeunit': result[4], 'description': result[5]}
return jsonify(data) return jsonify(data)
@ -129,7 +129,7 @@ def api_get_product_information(searchmethod, search):
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/brands", methods = ["GET"]) @app.route("/api/v1/brands", methods=["GET"])
def api_list_brands(): def api_list_brands():
"""Route to list all brands""" """Route to list all brands"""
session = Session() session = Session()
@ -142,20 +142,20 @@ def api_list_brands():
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/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):
"""Route to get information about a location""" """Route to get information about a location"""
session = Session() session = Session()
try: try:
data = session.query(Brand.name, Brand.description). \ data = session.query(Brand.name, Brand.description). \
filter(Brand.name == search).one() filter(Brand.name == search).one()
except NoResultFound: except NoResultFound:
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
data2 = {'name': data.name, 'description': data.description} data2 = {'name': data.name, 'description': data.description}
return jsonify(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():
"""Route to create a new location""" """Route to create a new location"""
session = Session() session = Session()
@ -178,7 +178,7 @@ def api_create_location():
return jsonify({'api_endpoint': f'/api/v1/location/name/{locname}'}) 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():
"""Route to create a new brand""" """Route to create a new brand"""
session = Session() session = Session()
@ -191,7 +191,8 @@ def api_create_brand():
else: else:
branddesc = None branddesc = None
ins = sqlalchemy.insert(Brand).values(name=brandname, description=branddesc) ins = sqlalchemy.insert(Brand).values(
name=brandname, description=branddesc)
print(f"ins is {ins}") print(f"ins is {ins}")
session.execute(ins) session.execute(ins)
print("executed ins") print("executed ins")
@ -201,7 +202,7 @@ def api_create_brand():
return jsonify({'api_endpoints': [f'/api/v1/brand/name/{brandname}']}) return jsonify({'api_endpoints': [f'/api/v1/brand/name/{brandname}']})
@app.route("/api/v1/create_product", methods = ["POST"]) @app.route("/api/v1/create_product", methods=["POST"])
def api_create_product(): def api_create_product():
"""Route to create a new product""" """Route to create a new product"""
session = Session() session = Session()
@ -215,9 +216,9 @@ def api_create_product():
desc = None desc = None
ins = sqlalchemy.insert(Product).values(upc=request.form['upc'], brand=request.form['brand'], ins = sqlalchemy.insert(Product).values(upc=request.form['upc'], brand=request.form['brand'],
name=request.form['name'], size=request.form['size'], name=request.form['name'], size=request.form['size'],
sizeunit=request.form['sizeunit'], sizeunit=request.form['sizeunit'],
description=desc) description=desc)
print(f"ins is {ins}") print(f"ins is {ins}")
session.execute(ins) session.execute(ins)
print("executed ins") print("executed ins")