diff --git a/.gitignore b/.gitignore index e21a501..5a579ff 100644 --- a/.gitignore +++ b/.gitignore @@ -129,5 +129,8 @@ dmypy.json # Pyre type checker .pyre/ +# VSCode custom files +.vscode/ + # InvMan custom postgres.ini diff --git a/.pylintrc b/.pylintrc index 55af2f3..ee8cf19 100644 --- a/.pylintrc +++ b/.pylintrc @@ -1,5 +1,5 @@ [TYPECHECK] -ignored-modules=flask_sqlalchemy +ignored-modules=flask, flask_api, sqlalchemy generated-members=session.* [MESSAGES CONTROL] diff --git a/libdb.py b/libdb.py index fe9eb76..1c26c14 100644 --- a/libdb.py +++ b/libdb.py @@ -6,7 +6,7 @@ Library for interacting with InvMan DB using SQLAlchemy """ 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.orm import sessionmaker @@ -53,3 +53,23 @@ class ProductQuantity(Base): name = Column(Text) quantity = Column(BigInteger) 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) diff --git a/webapi.py b/webapi.py index b81b2cd..618c6c0 100644 --- a/webapi.py +++ b/webapi.py @@ -6,11 +6,12 @@ Web API for InvMan """ # Imports -from flask import Flask, jsonify +from flask import Flask, jsonify, request from flask_api import status 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 @@ -58,156 +59,173 @@ def api_get_current_quantities(location): return jsonify({'error': 'NO_RESULT_FOUND'}), status.HTTP_404_NOT_FOUND -""" @app.route("/api/v1/location//quantity//", methods = ["GET"]) +@app.route("/api/v1/location//quantity//", methods = ["GET"]) def api_get_quantity_of_product_in_location(location, searchmethod, search): - db = get_db() - with db.cursor() as cur: - methodmap = {'upc': 'product_upc', 'name': 'name'} - if searchmethod in methodmap.keys(): - method = methodmap[searchmethod] - cur.execute("SELECT quantity FROM product_quantity WHERE location = %s AND {0} = %s".format(method), (location, search)) - data = cur.fetchall() - print("ran SELECT quantity FROM product_quantity WHERE location = %s AND {0} = %s".format(method)) - if len(data) <= 0: - print("no results returned, sending 404") - return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND - else: - data2 = data[0][0] - print("processed data:") - print(data2) - return jsonify(data2) + """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 """ + 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(): - db = get_db() - with db.cursor() as cur: - cur.execute("SELECT upc,brand,name,size,sizeunit,description FROM product") - data = cur.fetchall() - print("ran SELECT upc,brand,name,size,sizeunit,description FROM product") - if len(data) <= 0: - print("no results returned, sending 404") - return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND - else: - data2 = {} - for row in data: - data2[row[0]] = {'brand': row[1], 'name': row[2], 'size': row[3], 'sizeunit': row[4], 'description': row[5]} - print("processed data:") - print(data2) - return jsonify(data2) """ + """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"]) +@app.route("/api/v1/product//", methods = ["GET"]) def api_get_product_information(searchmethod, search): - db = get_db() - with db.cursor() as cur: - if searchmethod in ['name', 'upc']: - cur.execute("SELECT upc,brand,name,size,sizeunit,description FROM product WHERE {0} = %s".format(searchmethod), (search,)) - data = cur.fetchall() - print("ran SELECT upc,brand,name,size,sizeunit,description FROM product WHERE {0} = %s") - if len(data) <= 0: - print("no results returned, sending 404") - return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND - else: - row = data[0] + """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) - data2 = {} - data2['upc'] = row[0] - data2['brand'] = row[1] - data2['name'] = row[2] - data2['size'] = row[3] - data2['sizeunit'] = row[4] - data2['description'] = row[5] + 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) - print("processed data:") - print(data2) - return jsonify(data2) else: 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(): - db = get_db() - with db.cursor() as cur: - cur.execute("SELECT name,description FROM brand") - data = cur.fetchall() - print("ran SELECT name,description FROM brand") - if len(data) <= 0: - print("no results returned, sending 404") - return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND - else: - data2 = {} - for row in data: - data2[row[0]] = {'description': row[1]} - print("processed data:") - print(data2) - return jsonify(data2) """ + """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"]) +@app.route("/api/v1/brand/name/", methods = ["GET"]) def api_get_brand_by_name(search): - db = get_db() - with db.cursor() as cur: - cur.execute("SELECT name,description FROM brand WHERE name = %s", (search,)) - data = cur.fetchall() - print("ran SELECT name,description FROM brand WHERE name = %s") - if len(data) <= 0: - print("no results returned, sending 404") - return jsonify({'error': 'NO_RESULTS_RETURNED'}), status.HTTP_404_NOT_FOUND - else: - data2 = {} - for row in data: - data2[row[0]] = {'description': row[1]} - print("processed data:") - print(data2) - return jsonify(data2) """ + """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/create_location", methods = ["POST"]) +@app.route("/api/v1/create_location", methods = ["POST"]) def api_create_location(): - db = get_db() - with db.cursor() as cur: - 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 + """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 + + ins = sqlalchemy.insert(Location).values(name=locname, description=locdesc) + print(f"ins is {ins}") + session.execute(ins) + print("executed ins") + session.commit() + print("committed") - print(locname, locdesc) - cur.execute("INSERT INTO location (name, description) VALUES (%s, %s)", (locname, locdesc)) - print("ran INSERT INTO location (name, description) VALUES (%s, %s)") - db.commit() - print("committed changes to database") - return jsonify({'api_endpoint': '/api/v1/location/name/{0}'.format(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(): - db = get_db() - with db.cursor() as cur: - 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 + """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 + + ins = sqlalchemy.insert(Brand).values(name=brandname, description=branddesc) + print(f"ins is {ins}") + session.execute(ins) + print("executed ins") + session.commit() + print("committed") - print(f"creating brand with name {locname} and description {locdesc}") - cur.execute("INSERT INTO brand (name, description) VALUES (%s, %s)", (locname, locdesc)) - print("ran INSERT INTO brand (name, description) VALUES (%s, %s)") - db.commit() - print("committed changes to database") - return jsonify({'api_endpoint': '/api/v1/brand/name/{0}'.format(locname)}) """ + 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__":