396 lines
14 KiB
Python
396 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
InvMan Web API
|
|
|
|
Web API for InvMan
|
|
"""
|
|
|
|
# Imports
|
|
from flask import Flask, jsonify, request
|
|
from flask_api import status
|
|
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
from sqlalchemy.exc import IntegrityError
|
|
from libdb import Session, Location, ProductQuantity, Product, Brand, Unit, Purchase, Use
|
|
|
|
|
|
app = Flask(__name__) # app is the Flask app
|
|
|
|
|
|
@app.route("/api/v1/locations", methods=["GET"])
|
|
def get_locations():
|
|
"""Route to get locations"""
|
|
session = Session()
|
|
data = {}
|
|
try:
|
|
for name, desc in session.query(Location.name, Location.description):
|
|
data[name] = {'description': desc}
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
return jsonify(data)
|
|
|
|
|
|
@app.route("/api/v1/location/name/<search>", methods=["GET"])
|
|
def api_get_location_information(search):
|
|
"""Route to get information about a location"""
|
|
session = Session()
|
|
try:
|
|
data = session.query(Location.name, Location.description). \
|
|
filter(Location.name == search).one()
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
data2 = {}
|
|
data2['name'] = data.name
|
|
data2['description'] = data.description
|
|
return jsonify(data2)
|
|
|
|
|
|
@app.route("/api/v1/location/<location>/quantities", methods=["GET"])
|
|
def api_get_current_quantities(location):
|
|
"""Route to get quantities in a location"""
|
|
session = Session()
|
|
data = {}
|
|
try:
|
|
for upc, quantity in session.query(ProductQuantity.product_upc, ProductQuantity.quantity) \
|
|
.filter(ProductQuantity.location == location).all():
|
|
data[upc] = quantity
|
|
return jsonify(data)
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
@app.route("/api/v1/location/<location>/quantity/<searchmethod>/<search>", methods=["GET"])
|
|
def api_get_quantity_of_product_in_location(location, searchmethod, search):
|
|
"""Route to get the quantity of a product at a location"""
|
|
session = Session()
|
|
try:
|
|
if searchmethod == 'upc':
|
|
data = session.query(ProductQuantity.quantity).filter(
|
|
ProductQuantity.location == location,
|
|
ProductQuantity.product_upc == search).one()
|
|
return jsonify(data[0])
|
|
elif searchmethod == 'name':
|
|
data = session.query(ProductQuantity.quantity).filter(
|
|
ProductQuantity.location == location,
|
|
ProductQuantity.name == search).one()
|
|
return jsonify(data[0])
|
|
else:
|
|
print("invalid search method, sending 400")
|
|
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"])
|
|
def api_get_products():
|
|
"""Route to get a list of products"""
|
|
session = Session()
|
|
try:
|
|
data = {}
|
|
for upc, brand, name, size, sizeunit, description in session.query(
|
|
Product.upc, Product.brand, Product.name,
|
|
Product.size, Product.sizeunit, Product.description).all():
|
|
data[upc] = {'brand': brand, 'name': name, 'size': size,
|
|
'sizeunit': sizeunit, 'description': description}
|
|
return jsonify(data)
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
@app.route("/api/v1/product/<searchmethod>/<search>", methods=["GET"])
|
|
def api_get_product_information(searchmethod, search):
|
|
"""Route to get information about a product"""
|
|
session = Session()
|
|
try:
|
|
if searchmethod == 'name':
|
|
result = session.query(Product.upc, Product.brand, Product.name,
|
|
Product.size, Product.sizeunit, Product.description).filter(
|
|
Product.name == search
|
|
).one()
|
|
data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3],
|
|
'sizeunit': result[4], 'description': result[5]}
|
|
return jsonify(data)
|
|
|
|
if searchmethod == 'upc':
|
|
result = session.query(Product.upc, Product.brand, Product.name,
|
|
Product.size, Product.sizeunit, Product.description).filter(
|
|
Product.upc == search
|
|
).one()
|
|
data = {'upc': result[0], 'brand': result[1], 'name': result[2], 'size': result[3],
|
|
'sizeunit': result[4], 'description': result[5]}
|
|
return jsonify(data)
|
|
|
|
else:
|
|
print("invalid search method, sending 400")
|
|
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"])
|
|
def api_get_brands():
|
|
"""Route to get a list of all brands"""
|
|
session = Session()
|
|
try:
|
|
data = {}
|
|
for name, description in session.query(Brand.name, Brand.description).all():
|
|
data[name] = {'description': description}
|
|
return jsonify(data)
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
@app.route("/api/v1/brand/name/<search>", methods=["GET"])
|
|
def api_get_brand_by_name(search):
|
|
"""Route to get information about a location"""
|
|
session = Session()
|
|
try:
|
|
data = session.query(Brand.name, Brand.description).filter(
|
|
Brand.name == search).one()
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
data2 = {'name': data.name, 'description': data.description}
|
|
return jsonify(data2)
|
|
|
|
|
|
@app.route("/api/v1/units", methods=["GET"])
|
|
def api_get_units():
|
|
"""Route to get a list of all units"""
|
|
session = Session()
|
|
try:
|
|
data = {}
|
|
for name, description in session.query(Unit.name, Unit.description).all():
|
|
data[name] = {'description': description}
|
|
return jsonify(data)
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
@app.route("/api/v1/unit/name/<search>", methods=["GET"])
|
|
def api_get_unit_by_name(search):
|
|
"""Route to get information about a unit"""
|
|
session = Session()
|
|
try:
|
|
data = session.query(Unit.name, Unit.description).filter(
|
|
Unit.name == search).one()
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
data2 = {'name': data.name, 'description': data.description}
|
|
return jsonify(data2)
|
|
|
|
|
|
@app.route("/api/v1/purchases", methods=["GET"])
|
|
def api_get_purchases():
|
|
"""Route to get a list of all purchases"""
|
|
session = Session()
|
|
try:
|
|
data = {}
|
|
query = session.query(Purchase.id,
|
|
Purchase.product_upc, Purchase.quantity, Purchase.date,
|
|
Purchase.location)
|
|
for purchase_id, upc, quantity, date, location in query:
|
|
data[purchase_id] = {'upc': upc, 'quantity': quantity,
|
|
'date': str(date), 'location': location}
|
|
return jsonify(data)
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
@app.route("/api/v1/purchase/id/<search>", methods=["GET"])
|
|
def api_get_purchase_by_id(search):
|
|
"""Route to get information about a purchase"""
|
|
session = Session()
|
|
try:
|
|
data = session.query(Purchase.id, Purchase.product_upc, Purchase.quantity,
|
|
Purchase.date, Purchase.location).filter(Purchase.id == search).one()
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
data2 = {'id': data.id, 'upc': data.product_upc,
|
|
'quantity': data.quantity, 'date': str(data.date), 'location': data.location}
|
|
return jsonify(data2)
|
|
|
|
|
|
@app.route("/api/v1/uses", methods=["GET"])
|
|
def api_get_uses():
|
|
"""Route to get a list of all uses"""
|
|
session = Session()
|
|
try:
|
|
data = {}
|
|
query = session.query(Use.id,
|
|
Use.product_upc, Use.quantity, Use.date,
|
|
Use.location)
|
|
for use_id, upc, quantity, date, location in query:
|
|
data[use_id] = {'upc': upc, 'quantity': quantity,
|
|
'date': str(date), 'location': location}
|
|
return jsonify(data)
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
@app.route("/api/v1/use/id/<search>", methods=["GET"])
|
|
def api_get_use_by_id(search):
|
|
"""Route to get information about a use"""
|
|
session = Session()
|
|
try:
|
|
data = session.query(Use.id, Use.product_upc, Use.quantity,
|
|
Use.date, Use.location).filter(Use.id == search).one()
|
|
except NoResultFound:
|
|
return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND
|
|
data2 = {'id': data.id, 'upc': data.product_upc,
|
|
'quantity': data.quantity, 'date': str(data.date), 'location': data.location}
|
|
return jsonify(data2)
|
|
|
|
|
|
@app.route("/api/v1/create_location", methods=["POST"])
|
|
def api_create_location():
|
|
"""Route to create a new location"""
|
|
session = Session()
|
|
if "name" not in request.form:
|
|
return jsonify({'error': 'NO_NAME_PROVIDED'}), status.HTTP_400_BAD_REQUEST
|
|
else:
|
|
locname = request.form['name']
|
|
if "description" in request.form:
|
|
locdesc = request.form['description']
|
|
else:
|
|
locdesc = None
|
|
|
|
newlocation = Location(name=locname, description=locdesc)
|
|
session.add(newlocation)
|
|
print("added newlocation")
|
|
try:
|
|
session.commit()
|
|
print("committed")
|
|
return jsonify({'api_endpoints': f'/api/v1/location/name/{locname}'})
|
|
except IntegrityError:
|
|
session.rollback()
|
|
return jsonify({'error': 'INTEGRITY_ERROR'})
|
|
|
|
|
|
@app.route("/api/v1/create_brand", methods=["POST"])
|
|
def api_create_brand():
|
|
"""Route to create a new brand"""
|
|
session = Session()
|
|
if "name" not in request.form:
|
|
return jsonify({'error': 'NO_NAME_PROVIDED'}), status.HTTP_400_BAD_REQUEST
|
|
else:
|
|
brandname = request.form['name']
|
|
if "description" in request.form:
|
|
branddesc = request.form['description']
|
|
else:
|
|
branddesc = None
|
|
|
|
newbrand = Brand(name=brandname, description=branddesc)
|
|
session.add(newbrand)
|
|
print("added newbrand")
|
|
try:
|
|
session.commit()
|
|
print("committed")
|
|
return jsonify({'api_endpoints': [f'/api/v1/brand/name/{brandname}']})
|
|
except IntegrityError:
|
|
session.rollback()
|
|
return jsonify({'error': 'INTEGRITY_ERROR'})
|
|
|
|
|
|
@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
|
|
|
|
newproduct = Product(upc=request.form['upc'], brand=request.form['brand'],
|
|
name=request.form['name'], size=request.form['size'],
|
|
sizeunit=request.form['sizeunit'],
|
|
description=desc)
|
|
session.add(newproduct)
|
|
print("added newproduct")
|
|
try:
|
|
session.commit()
|
|
print("committed")
|
|
return jsonify({'api_endpoints': [f'/api/v1/product/upc/{request.form["upc"]}',
|
|
f'/api/v1/product/name/{request.form["name"]}']})
|
|
except IntegrityError:
|
|
session.rollback()
|
|
return jsonify({'error': 'INTEGRITY_ERROR'})
|
|
|
|
|
|
@app.route("/api/v1/create_unit", methods=["POST"])
|
|
def api_create_unit():
|
|
"""Route to create a new unit"""
|
|
session = Session()
|
|
if "name" not in request.form:
|
|
return jsonify({'error': 'NO_NAME_PROVIDED'}), status.HTTP_400_BAD_REQUEST
|
|
else:
|
|
unitname = request.form['name']
|
|
if "description" in request.form:
|
|
unitdesc = request.form['description']
|
|
else:
|
|
unitdesc = None
|
|
|
|
newunit = Unit(name=unitname, description=unitdesc)
|
|
session.add(newunit)
|
|
print("added newunit")
|
|
try:
|
|
session.commit()
|
|
print("committed")
|
|
return jsonify({'api_endpoints': [f'/api/v1/unit/name/{unitname}']})
|
|
except IntegrityError:
|
|
session.rollback()
|
|
return jsonify({'error': 'INTEGRITY_ERROR'})
|
|
|
|
|
|
@app.route("/api/v1/create_purchase", methods=["POST"])
|
|
def api_create_purchase():
|
|
"""Route to create a new purchase"""
|
|
session = Session()
|
|
reqfields = ['upc', 'quantity', 'date', 'location']
|
|
for field in reqfields:
|
|
if field not in request.form:
|
|
return jsonify({'error': f'NO_{field.upper()}_PROVIDED'}), status.HTTP_400_BAD_REQUEST
|
|
|
|
newpurchase = Purchase(product_upc=request.form['upc'], quantity=request.form['quantity'],
|
|
date=request.form['date'], location=request.form['location'])
|
|
session.add(newpurchase)
|
|
print("added newpurchase")
|
|
try:
|
|
session.commit()
|
|
print("committed")
|
|
return jsonify({'api_endpoints': [f'/api/v1/purchase/id/{newpurchase.id}']})
|
|
except IntegrityError:
|
|
session.rollback()
|
|
return jsonify({'error': 'INTEGRITY_ERROR'})
|
|
|
|
|
|
@app.route("/api/v1/create_use", methods=["POST"])
|
|
def api_create_use():
|
|
"""Route to create a new use"""
|
|
session = Session()
|
|
reqfields = ['upc', 'quantity', 'date', 'location']
|
|
for field in reqfields:
|
|
if field not in request.form:
|
|
return jsonify({'error': f'NO_{field.upper()}_PROVIDED'}), status.HTTP_400_BAD_REQUEST
|
|
|
|
newuse = Use(product_upc=request.form['upc'], quantity=request.form['quantity'],
|
|
date=request.form['date'], location=request.form['location'])
|
|
session.add(newuse)
|
|
print("added newuse")
|
|
try:
|
|
session.commit()
|
|
print("committed")
|
|
return jsonify({'api_endpoints': [f'/api/v1/use/id/{newuse.id}']})
|
|
except IntegrityError:
|
|
session.rollback()
|
|
return jsonify({'error': 'INTEGRITY_ERROR'})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Run with `flask` or a WSGI server!")
|