#!/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 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/", 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//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//quantity//", 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//", methods=["GET"]) def api_get_product_information(searchmethod, search): """Route to get a 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_list_brands(): """Route to list 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/", 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_list_units(): """Route to list 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/", 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_list_purchases(): """Route to list 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/", 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_list_uses(): """Route to list 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/", 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") session.commit() print("committed") return jsonify({'api_endpoint': f'/api/v1/location/name/{locname}'}) @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") session.commit() 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 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") session.commit() print("committed") return jsonify({'api_endpoints': [f'/api/v1/product/upc/{request.form["upc"]}', f'/api/v1/product/name/{request.form["name"]}']}) @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") session.commit() print("committed") return jsonify({'api_endpoints': [f'/api/v1/unit/name/{unitname}']}) @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") session.commit() print("committed") return jsonify({'api_endpoints': [f'/api/v1/purchase/id/{newpurchase.id}']}) @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") session.commit() print("committed") return jsonify({'api_endpoints': [f'/api/v1/use/id/{newuse.id}']}) if __name__ == "__main__": print("Run with `flask` or a WSGI server!")