jadowyne/pantry-track
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Finished clean up of posting new item with api

Browse Source
This commit is contained in:
Jadowyne Ulve 2025-05-31 18:24:32 -05:00
parent aec8f85a4d
commit a006668ac1
3 changed files with 119 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
application/items/database_items.py
View File

@ -510,14 +510,33 @@ def insertCostLayersTuple(site, payload, convert=True, conn=None):
except Exception as error:
raise postsqldb.DatabaseError(error, payload, sql)
def insertItemLocationsTuple(conn, site, payload, convert=True):
def insertItemLocationsTuple(site, payload, convert=True, conn=None):
"""insert payload into item_locations table for site
Args:
conn (_T_connector@connect): Postgresql Connector
site (str):
payload (tuple): (part_id[int], location_id[int], quantity_on_hand[float], cost_layers[lst2pgarr])
convert (bool, optional): Determines if to return tuple as dictionary. Defaults to False.
Raises:
DatabaseError:
Returns:
tuple or dict: inserted tuple
"""
location = ()
self_conn = False
database_config = config.config()
with open(f"application/items/sql/insertItemLocationsTuple.sql", "r+") as file:
sql = file.read().replace("%%site_name%%", site)
try:
conn = psycopg2.connect(**database_config)
conn.autocommit = False
if not conn:
database_config = config.config()
conn = psycopg2.connect(**database_config)
conn.autocommit = True
self_conn = True
with conn.cursor() as cur:
cur.execute(sql, payload)
rows = cur.fetchone()
@ -525,11 +544,16 @@ def insertItemLocationsTuple(conn, site, payload, convert=True):
location = postsqldb.tupleDictionaryFactory(cur.description, rows)
elif rows and not convert:
location = rows
return location, conn
if self_conn:
conn.commit()
conn.close()
return location
except Exception as error:
raise postsqldb.DatabaseError(error, payload, sql)
def insertLogisticsInfoTuple(conn, site, payload, convert=False):
def insertLogisticsInfoTuple(site, payload, convert=True, conn=None):
"""insert payload into logistics_info table for site
Args:
@ -546,9 +570,17 @@ def insertLogisticsInfoTuple(conn, site, payload, convert=False):
tuple or dict: inserted tuple
"""
logistics_info = ()
self_conn = False
with open(f"application/items/sql/insertLogisticsInfoTuple.sql", "r+") as file:
sql = file.read().replace("%%site_name%%", site)
try:
if not conn:
database_config = config.config()
conn = psycopg2.connect(**database_config)
conn.autocommit = True
self_conn = True
with conn.cursor() as cur:
cur.execute(sql, payload)
rows = cur.fetchone()
@ -556,12 +588,17 @@ def insertLogisticsInfoTuple(conn, site, payload, convert=False):
logistics_info = postsqldb.tupleDictionaryFactory(cur.description, rows)
elif rows and not convert:
logistics_info = rows
if self_conn:
conn.commit()
conn.close()
return logistics_info
except Exception as error:
raise postsqldb.DatabaseError(error, payload, sql)
return logistics_info
def insertItemInfoTuple(conn, site, payload, convert=False):
def insertItemInfoTuple(site, payload, convert=True, conn=None):
"""inserts payload into the item_info table of site
Args:
@ -578,9 +615,16 @@ def insertItemInfoTuple(conn, site, payload, convert=False):
tuple or dict: inserted tuple
"""
item_info = ()
self_conn = False
with open(f"application/items/sql/insertItemInfoTuple.sql", "r+") as file:
sql = file.read().replace("%%site_name%%", site)
try:
if not conn:
database_config = config.config()
conn = psycopg2.connect(**database_config)
conn.autocommit = True
self_conn = True
with conn.cursor() as cur:
cur.execute(sql, payload)
rows = cur.fetchone()
@ -588,11 +632,15 @@ def insertItemInfoTuple(conn, site, payload, convert=False):
item_info = postsqldb.tupleDictionaryFactory(cur.description, rows)
elif rows and not convert:
item_info = rows
if self_conn:
conn.commit()
conn.close()
return item_info
except Exception as error:
raise postsqldb.DatabaseError(error, payload, sql)
return item_info
def insertFoodInfoTuple(conn, site, payload, convert=False):
def insertFoodInfoTuple(site, payload, convert=True, conn=None):
"""insert payload into food_info table for site
Args:
@ -608,9 +656,16 @@ def insertFoodInfoTuple(conn, site, payload, convert=False):
tuple or dict: inserted tuple
"""
food_info = ()
self_conn = False
with open(f"application/items/sql/insertFoodInfoTuple.sql", "r+") as file:
sql = file.read().replace("%%site_name%%", site)
try:
if not conn:
database_config = config.config()
conn = psycopg2.connect(**database_config)
conn.autocommit = True
self_conn = True
with conn.cursor() as cur:
cur.execute(sql, payload)
rows = cur.fetchone()
@ -618,11 +673,16 @@ def insertFoodInfoTuple(conn, site, payload, convert=False):
food_info = postsqldb.tupleDictionaryFactory(cur.description, rows)
elif rows and not convert:
food_info = rows
if self_conn:
conn.commit()
conn.close()
return food_info
except Exception as error:
raise postsqldb.DatabaseError(error, payload, sql)
return food_info
def insertItemTuple(conn, site, payload, convert=False):
def insertItemTuple(site, payload, convert=True, conn=None):
"""insert payload into items table for site
Args:
@ -640,9 +700,16 @@ def insertItemTuple(conn, site, payload, convert=False):
tuple or dict: inserted tuple
"""
item = ()
self_conn = False
with open(f"application/items/sql/insertItemTuple.sql", "r+") as file:
sql = file.read().replace("%%site_name%%", site)
try:
if not conn:
database_config = config.config()
conn = psycopg2.connect(**database_config)
conn.autocommit = True
self_conn = True
with conn.cursor() as cur:
cur.execute(sql, payload)
rows = cur.fetchone()
@ -650,40 +717,14 @@ def insertItemTuple(conn, site, payload, convert=False):
item = postsqldb.tupleDictionaryFactory(cur.description, rows)
elif rows and not convert:
item = rows
if self_conn:
conn.commit()
conn.close()
return item
except Exception as error:
raise postsqldb.DatabaseError(error, payload, sql)
return item
def insertItemLocationsTuple(conn, site, payload, convert=False):
"""insert payload into item_locations table for site
Args:
conn (_T_connector@connect): Postgresql Connector
site (str):
payload (tuple): (part_id[int], location_id[int], quantity_on_hand[float], cost_layers[lst2pgarr])
convert (bool, optional): Determines if to return tuple as dictionary. Defaults to False.
Raises:
DatabaseError:
Returns:
tuple or dict: inserted tuple
"""
location = ()
with open(f"application/items/sql/insertItemLocationsTuple.sql", "r+") as file:
sql = file.read().replace("%%site_name%%", site)
try:
with conn.cursor() as cur:
cur.execute(sql, payload)
rows = cur.fetchone()
if rows and convert:
location = postsqldb.tupleDictionaryFactory(cur.description, rows)
elif rows and not convert:
location = rows
except Exception as error:
raise postsqldb.DatabaseError(error, payload, sql)
return location
def selectItemLocationsTuple(site_name, payload, convert=True):
"""select a single tuple from ItemLocations table for site_name

31
application/items/items_API.py
View File

@ -417,7 +417,6 @@ def getBrands():
return jsonify({'brands': brands, 'endpage': math.ceil(count/limit), 'error': False, 'message': f''})
return jsonify({'brands': brands, 'endpage': math.ceil(count/limit), 'error': True, 'message': f'method {request.method} is not allowed.'})
@items_api.route('/updateItem', methods=['POST'])
@login_required
def updateItem():
@ -493,7 +492,6 @@ def updateItemLink():
return jsonify({'error':False, 'message': "Linked Item was updated successfully"})
return jsonify({'error': True, 'message': f"method {request.method} not allowed."})
@items_api.route('/getPossibleLocations', methods=["GET"])
@login_required
def getPossibleLocations():
@ -603,13 +601,40 @@ def addLinkedItem():
@items_api.route('/addBlankItem', methods=["POST"])
def addBlankItem():
""" POST new Blank item to the system given a barcode, item_name, subtype
---
parameters:
- in: query
name: barcode
schema:
type: string
default: 1
required: true
description: barcode for the item
- in: query
name: item_name
schema:
type: string
default: 1
required: true
description: name of the blank item
- in: query
name: subtype
schema:
type: string
default: 1
required: true
description: type of item this is categorized to be.
responses:
200:
description: Item added successfully.
"""
if request.method == "POST":
data = {
'barcode': request.get_json()['barcode'],
'name': request.get_json()['name'],
'subtype': request.get_json()['subtype']
}
database_config = config()
site_name = session['selected_site']
user_id = session['user_id']

10
application/items/items_processes.py
View File

@ -42,9 +42,9 @@ def postNewBlankItem(site_name: str, user_id: int, data: dict, conn=None):
brand_id = 1
logistics_info = database_items.insertLogisticsInfoTuple(conn, site_name, logistics_info.payload(), convert=True)
item_info = database_items.insertItemInfoTuple(conn, site_name, item_info.payload(), convert=True)
food_info = database_items.insertFoodInfoTuple(conn, site_name, food_info.payload(), convert=True)
logistics_info = database_items.insertLogisticsInfoTuple(site_name, logistics_info.payload(), conn=conn)
item_info = database_items.insertItemInfoTuple(site_name, item_info.payload(), conn=conn)
food_info = database_items.insertFoodInfoTuple(site_name, food_info.payload(), conn=conn)
name = data['name']
name = name.replace("'", "@&apostraphe&")
@ -66,7 +66,7 @@ def postNewBlankItem(site_name: str, user_id: int, data: dict, conn=None):
search_string=search_string
)
item = database_items.insertItemTuple(conn, site_name, item.payload(), convert=True)
item = database_items.insertItemTuple(site_name, item.payload(), conn=conn)
with conn.cursor() as cur:
cur.execute(f"SELECT id FROM {site_name}_locations WHERE uuid=%s;", (uuid, ))
@ -74,7 +74,7 @@ def postNewBlankItem(site_name: str, user_id: int, data: dict, conn=None):
dbPayloads.ItemLocationPayload
item_location = dbPayloads.ItemLocationPayload(item['id'], location_id)
database_items.insertItemLocationsTuple(conn, site_name, item_location.payload())
database_items.insertItemLocationsTuple(site_name, item_location.payload(), conn=conn)
creation_tuple = dbPayloads.TransactionPayload(