Save work for 9/28/2020
This commit is contained in:
parent
2cc2a3917a
commit
fe898cde00
GPG Key ID:
18089E4E3CCF1D3A
|
|
@ -130,4 +130,4 @@ dmypy.json
|
|||
.pyre/
|
||||
|
||||
# InvMan custom
|
||||
inventory.json
|
||||
postgres.ini
|
||||
|
|
|
|||
|
|
@ -0,0 +1,54 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
# Imports
|
||||
import psycopg2
|
||||
from configparser import ConfigParser
|
||||
|
||||
|
||||
class Database:
|
||||
def __init__(self):
|
||||
filename = "postgres.ini"
|
||||
section = "postgresql"
|
||||
parser = ConfigParser()
|
||||
parser.read(filename)
|
||||
|
||||
config = {}
|
||||
if parser.has_section(section):
|
||||
params = parser.items(section)
|
||||
for param in params:
|
||||
config[param[0]] = param[1]
|
||||
else:
|
||||
raise Exception('Section {0} not found in the {1} file'.format(section, filename))
|
||||
|
||||
self.host = config["host"]
|
||||
self.user = config["user"]
|
||||
self.password = config["password"]
|
||||
self.port = 5432
|
||||
self.dbname = config["database"]
|
||||
self.conn = None
|
||||
|
||||
|
||||
def connect(self):
|
||||
if self.conn is None:
|
||||
try:
|
||||
self.conn = psycopg2.connect(host=self.host, user=self.user, password=self.password, port=self.port, dbname=self.dbname)
|
||||
except psycopg2.DatabaseError as e:
|
||||
print(e)
|
||||
raise e
|
||||
finally:
|
||||
print("Connection opened successfully.")
|
||||
|
||||
|
||||
def runcmd_unsafe(self, cmd):
|
||||
self.connect()
|
||||
with self.conn.cursor() as cur:
|
||||
cur.execute(cmd)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
def list_types(self):
|
||||
self.connect()
|
||||
with self.conn.cursor() as cur:
|
||||
cur.execute("SELECT name FROM types")
|
||||
types = [row for row in cur.fetchall()]
|
||||
return types
|
||||
|
|
@ -1,2 +1,4 @@
|
|||
flask
|
||||
flask_api
|
||||
Flask
|
||||
Flask_API
|
||||
psycopg2
|
||||
sqlalchemy
|
||||
|
|
|
|||
92
storage.py
92
storage.py
|
|
@ -1,92 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
InvMan Storage Module
|
||||
"""
|
||||
|
||||
# Imports
|
||||
from flask import jsonify
|
||||
from flask_api import status
|
||||
import json
|
||||
|
||||
|
||||
class AlreadyExistsError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def load_raw_json():
|
||||
with open("inventory.json", "r") as f:
|
||||
print("Opened inventory.json")
|
||||
raw_json = json.loads(f.read())
|
||||
print("Converted JSON to dictionary")
|
||||
return raw_json
|
||||
|
||||
|
||||
def write_raw_json(writedict):
|
||||
with open("inventory.json", "w+") as f:
|
||||
print("Opened inventory.json for writing")
|
||||
f.write(json.dumps(writedict))
|
||||
print("Wrote dictionary as JSON to inventory.json")
|
||||
|
||||
|
||||
def get_locations():
|
||||
return list(load_raw_json().keys())
|
||||
|
||||
|
||||
def get_items(location):
|
||||
return load_raw_json()[location]
|
||||
|
||||
|
||||
def init_location(location):
|
||||
rawitems = load_raw_json()
|
||||
if location not in rawitems.keys():
|
||||
rawitems.update({location: {}})
|
||||
write_raw_json(rawitems)
|
||||
else:
|
||||
raise AlreadyExistsError
|
||||
|
||||
|
||||
def init_item(location, item):
|
||||
rawitems = load_raw_json()
|
||||
if item not in rawitems[location].keys():
|
||||
rawitems[location].update({item: {}})
|
||||
write_raw_json(rawitems)
|
||||
else:
|
||||
raise AlreadyExistsError
|
||||
|
||||
|
||||
def init_brand(location, item, brand):
|
||||
rawitems = load_raw_json()
|
||||
if item not in rawitems[location].keys():
|
||||
init_item(location, item)
|
||||
rawitems = load_raw_json()
|
||||
|
||||
if brand not in rawitems[location][item].keys():
|
||||
rawitems[location][item].update({brand: {}})
|
||||
write_raw_json(rawitems)
|
||||
else:
|
||||
raise AlreadyExistsError
|
||||
|
||||
|
||||
def increment_amount(location, item, brand, amount):
|
||||
rawitems = load_raw_json()
|
||||
newamt = rawitems[location][item][brand][amount] + int(amount)
|
||||
rawitems[location][item][brand].update({"amount": newamt})
|
||||
write_raw_json(rawitems)
|
||||
|
||||
|
||||
def rm_brand_field(location, item, brand, field):
|
||||
rawitems = load_raw_json()
|
||||
rawitems[location][item][brand].pop(field)
|
||||
write_raw_json(rawitems)
|
||||
|
||||
|
||||
def rm_brand(location, item, brand):
|
||||
rawitems = load_raw_json()
|
||||
rawitems[location][item].pop(brand)
|
||||
write_raw_json(rawitems)
|
||||
|
||||
|
||||
def rm_item(location, item):
|
||||
rawitems = load_raw_json()
|
||||
rawitems[location].pop(item)
|
||||
write_raw_json(rawitems)
|
||||
|
|
@ -0,0 +1,54 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
libinv
|
||||
|
||||
Library for interacting with inventory database.
|
||||
"""
|
||||
|
||||
# Imports
|
||||
import psycopg2
|
||||
from configparser import ConfigParser
|
||||
|
||||
def config(filename="postgres.ini", section="postgresql"):
|
||||
parser = ConfigParser()
|
||||
parser.read(filename)
|
||||
|
||||
db = {}
|
||||
if parser.has_section(section):
|
||||
params = parser.items(section)
|
||||
for param in params:
|
||||
db[param[0]] = param[1]
|
||||
else:
|
||||
raise Exception('Section {0} not found in the {1} file'.format(section, filename))
|
||||
|
||||
return db
|
||||
|
||||
|
||||
def connect():
|
||||
""" Connect to the PostgreSQL database server """
|
||||
conn = None
|
||||
try:
|
||||
params = config()
|
||||
|
||||
print("Connecting to the PostgreSQL database...")
|
||||
conn = psycopg2.connect(**params)
|
||||
|
||||
cur = conn.cursor()
|
||||
|
||||
# execute a statement
|
||||
print("Postgres version:")
|
||||
cur.execute("SELECT version()")
|
||||
|
||||
# display the database version
|
||||
db_version = cur.fetchone()
|
||||
print(db_version)
|
||||
except (Exception, psycopg2.DatabaseError) as e:
|
||||
print(e)
|
||||
finally:
|
||||
if conn is not None:
|
||||
conn.close()
|
||||
print("Database connection closed.")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
connect()
|
||||
176
webapi.py
176
webapi.py
|
|
@ -6,115 +6,119 @@ InvMan
|
|||
# Imports
|
||||
from flask import Flask, request, jsonify, abort
|
||||
from flask_api import status
|
||||
import psycopg2
|
||||
|
||||
import storage
|
||||
|
||||
app = Flask(__name__) # app is the Flask app
|
||||
|
||||
|
||||
@app.route("/api/v1/locations", methods = ["GET"])
|
||||
def api_get_locations():
|
||||
return jsonify(storage.get_locations())
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/items", methods = ["GET"])
|
||||
def api_get_items(location):
|
||||
try:
|
||||
items = storage.get_items(location)
|
||||
print("Got items at {0}".format(location))
|
||||
return jsonify(items)
|
||||
except KeyError:
|
||||
print("KeyError, returning 404")
|
||||
return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
# @app.route("/api/v1/locations", methods = ["GET"])
|
||||
# def api_get_locations():
|
||||
# return jsonify(storage.get_locations())
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/item/<item>", methods = ["GET", "DELETE"])
|
||||
def api_get_item(location, item):
|
||||
if request.method == "GET":
|
||||
try:
|
||||
itemresp = storage.get_items(location)[item.lower()]
|
||||
print("Got {0}".format(item))
|
||||
return jsonify(itemresp)
|
||||
except KeyError:
|
||||
print("KeyError, returning 404")
|
||||
return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
elif request.method == "DELETE":
|
||||
try:
|
||||
storage.rm_item(location, item)
|
||||
print("Delete item {0} at {1}".format(item, location))
|
||||
return jsonify({'status': 'success'})
|
||||
except KeyError:
|
||||
print("KeyError, returning 404")
|
||||
return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
# @app.route("/<location>/api/v1/items", methods = ["GET"])
|
||||
# def api_get_items(location):
|
||||
# try:
|
||||
# items = storage.get_items(location)
|
||||
# print("Got items at {0}".format(location))
|
||||
# return jsonify(items)
|
||||
# except KeyError:
|
||||
# print("KeyError, returning 404")
|
||||
# return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/item/<item>/<brand>", methods = ["GET", "DELETE"])
|
||||
def api_item_brand(location, item, brand):
|
||||
if request.method == "GET":
|
||||
try:
|
||||
itemresp = storage.get_items(location)[item.lower()][brand.lower()]
|
||||
print("Got {0} of brand {1} at {2}".format(item, brand, location))
|
||||
return jsonify(itemresp)
|
||||
except KeyError:
|
||||
print("KeyError, returning 404")
|
||||
return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
# @app.route("/<location>/api/v1/item/<item>", methods = ["GET", "DELETE"])
|
||||
# def api_get_item(location, item):
|
||||
# if request.method == "GET":
|
||||
# try:
|
||||
# itemresp = storage.get_items(location)[item.lower()]
|
||||
# print("Got {0}".format(item))
|
||||
# return jsonify(itemresp)
|
||||
# except KeyError:
|
||||
# print("KeyError, returning 404")
|
||||
# return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
elif request.method == "DELETE":
|
||||
try:
|
||||
storage.rm_brand(location, item, brand)
|
||||
print("Deleted {0} of brand {1} at {2}".format(item, brand, location))
|
||||
return jsonify({'status': 'success'})
|
||||
except KeyError:
|
||||
print("KeyError, returning 404")
|
||||
return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
# elif request.method == "DELETE":
|
||||
# try:
|
||||
# storage.rm_item(location, item)
|
||||
# print("Delete item {0} at {1}".format(item, location))
|
||||
# return jsonify({'status': 'success'})
|
||||
# except KeyError:
|
||||
# print("KeyError, returning 404")
|
||||
# return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/item/<item>/<brand>/<field>", methods = ["DELETE"])
|
||||
def api_rm_brand_field(location, item, brand, field):
|
||||
try:
|
||||
storage.rm_brand_field(location, item, brand, field)
|
||||
print("Deleted field {0} from brand {1} of item {2} at {3}".format(field, brand, item, location))
|
||||
return jsonify({'status': 'success'})
|
||||
except KeyError:
|
||||
print("KeyError, returning 404")
|
||||
return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
# @app.route("/<location>/api/v1/item/<item>/<brand>", methods = ["GET", "DELETE"])
|
||||
# def api_item_brand(location, item, brand):
|
||||
# if request.method == "GET":
|
||||
# try:
|
||||
# itemresp = storage.get_items(location)[item.lower()][brand.lower()]
|
||||
# print("Got {0} of brand {1} at {2}".format(item, brand, location))
|
||||
# return jsonify(itemresp)
|
||||
# except KeyError:
|
||||
# print("KeyError, returning 404")
|
||||
# return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
# elif request.method == "DELETE":
|
||||
# try:
|
||||
# storage.rm_brand(location, item, brand)
|
||||
# print("Deleted {0} of brand {1} at {2}".format(item, brand, location))
|
||||
# return jsonify({'status': 'success'})
|
||||
# except KeyError:
|
||||
# print("KeyError, returning 404")
|
||||
# return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/item/<item>/<brand>/new_brand", methods = ["POST"])
|
||||
def api_new_brand(location, item, brand):
|
||||
try:
|
||||
storage.init_brand(location, item, brand)
|
||||
print("Created brand {0} under item {1} at {2}".format(brand, item, location))
|
||||
return jsonify({'status': 'success'})
|
||||
except storage.AlreadyExistsError:
|
||||
return jsonify({'status': 'error', 'error': 'BRAND_ALREADY_EXISTS'}), status.HTTP_403_FORBIDDEN
|
||||
# @app.route("/<location>/api/v1/item/<item>/<brand>/<field>", methods = ["DELETE"])
|
||||
# def api_rm_brand_field(location, item, brand, field):
|
||||
# try:
|
||||
# storage.rm_brand_field(location, item, brand, field)
|
||||
# print("Deleted field {0} from brand {1} of item {2} at {3}".format(field, brand, item, location))
|
||||
# return jsonify({'status': 'success'})
|
||||
# except KeyError:
|
||||
# print("KeyError, returning 404")
|
||||
# return jsonify({'status': 'error', 'error': 'NOT_FOUND'}), status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/item/<item>/new_item", methods = ["POST"])
|
||||
def api_new_item(location, item):
|
||||
try:
|
||||
storage.init_item(location, item)
|
||||
print("Created item {0} at {1}".format(item, location))
|
||||
return jsonify({'status': 'success'})
|
||||
except storage.AlreadyExistsError:
|
||||
return jsonify({'status': 'error', 'error': 'ITEM_ALREADY_EXISTS'}), status.HTTP_403_FORBIDDEN
|
||||
# @app.route("/<location>/api/v1/item/<item>/<brand>/new_brand", methods = ["POST"])
|
||||
# def api_new_brand(location, item, brand):
|
||||
# try:
|
||||
# storage.init_brand(location, item, brand)
|
||||
# print("Created brand {0} under item {1} at {2}".format(brand, item, location))
|
||||
# return jsonify({'status': 'success'})
|
||||
# except storage.AlreadyExistsError:
|
||||
# return jsonify({'status': 'error', 'error': 'BRAND_ALREADY_EXISTS'}), status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/new_location", methods = ["POST"])
|
||||
def api_new_location(location):
|
||||
try:
|
||||
storage.init_location(location)
|
||||
print("Created location {0}".format(location))
|
||||
return jsonify({'status': 'success'})
|
||||
except storage.AlreadyExistsError:
|
||||
return jsonify({'status': 'error', 'error': 'ITEM_ALREADY_EXISTS'}), status.HTTP_403_FORBIDDEN
|
||||
# @app.route("/<location>/api/v1/item/<item>/new_item", methods = ["POST"])
|
||||
# def api_new_item(location, item):
|
||||
# try:
|
||||
# storage.init_item(location, item)
|
||||
# print("Created item {0} at {1}".format(item, location))
|
||||
# return jsonify({'status': 'success'})
|
||||
# except storage.AlreadyExistsError:
|
||||
# return jsonify({'status': 'error', 'error': 'ITEM_ALREADY_EXISTS'}), status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
@app.route("/<location>/api/v1/item/<item>/<brand>/add_amount/<amount>", methods = ["POST"])
|
||||
def api_add_amount(location, item, brand, amount):
|
||||
storage.increment_amount(location, item, brand, amount)
|
||||
return jsonify({'status': 'success'})
|
||||
# @app.route("/<location>/api/v1/new_location", methods = ["POST"])
|
||||
# def api_new_location(location):
|
||||
# try:
|
||||
# storage.init_location(location)
|
||||
# print("Created location {0}".format(location))
|
||||
# return jsonify({'status': 'success'})
|
||||
# except storage.AlreadyExistsError:
|
||||
# return jsonify({'status': 'error', 'error': 'ITEM_ALREADY_EXISTS'}), status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
# @app.route("/<location>/api/v1/item/<item>/<brand>/add_amount/<amount>", methods = ["POST"])
|
||||
# def api_add_amount(location, item, brand, amount):
|
||||
# storage.increment_amount(location, item, brand, amount)
|
||||
# return jsonify({'status': 'success'})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
Loading…
Reference in New Issue