From 009e3b6b1d7aa75809f1a4e178f7b0503d08836c Mon Sep 17 00:00:00 2001 From: Jadowyne Ulve Date: Sat, 31 May 2025 19:26:20 -0500 Subject: [PATCH] Migrated SKUPrefix to the new api design --- application/database_payloads.py | 15 ++ application/items/database_items.py | 209 +++++++++++------- application/items/items_API.py | 47 +++- application/items/items_processes.py | 15 +- .../items/sql/insertSKUPrefixTuple.sql | 4 + 5 files changed, 197 insertions(+), 93 deletions(-) create mode 100644 application/items/sql/insertSKUPrefixTuple.sql diff --git a/application/database_payloads.py b/application/database_payloads.py index ba7907e..ac03ff1 100644 --- a/application/database_payloads.py +++ b/application/database_payloads.py @@ -396,6 +396,21 @@ class ItemLocationPayload: lst2pgarr(self.cost_layers) ) +@dataclass +class SKUPrefixPayload: + __slots__ = ('uuid', 'name', 'description') + + uuid: str + name: str + description: str + + def payload(self): + return ( + self.uuid, + self.name, + self.description + ) + @dataclass class SiteManager: site_name: str diff --git a/application/items/database_items.py b/application/items/database_items.py index 8f59929..b373edb 100644 --- a/application/items/database_items.py +++ b/application/items/database_items.py @@ -4,6 +4,16 @@ import psycopg2 import datetime def getTransactions(site:str, payload: tuple, convert:bool=True): + """ Page through a sites Transactions by passing a logistics id, limit, and offset through a payload + + Args: + site (str): _description_ + payload (tuple): (logistics_id, limit, offset) + convert (bool, optional): _description_. Defaults to True. + + Returns: + _type_: _description_ + """ database_config = config.config() sql = f"SELECT * FROM {site}_transactions WHERE logistics_info_id=%s LIMIT %s OFFSET %s;" sql_count = f"SELECT COUNT(*) FROM {site}_transactions WHERE logistics_info_id=%s;" @@ -198,6 +208,93 @@ def getZone(site:str, payload:tuple, convert:bool=True): except Exception as error: raise postsqldb.DatabaseError(error, payload, sql) +def selectItemLocationsTuple(site_name, payload, convert=True): + """select a single tuple from ItemLocations table for site_name + + Args: + conn (_T_connector@connect): + site_name (str): + payload (tuple): [item_id, location_id] + convert (bool): defaults to False, used to determine return of tuple/dict + + Returns: + tuple: the row that was returned from the table + """ + item_locations = () + database_config = config.config() + select_item_location_sql = f"SELECT * FROM {site_name}_item_locations WHERE part_id = %s AND location_id = %s;" + try: + with psycopg2.connect(**database_config) as conn: + with conn.cursor() as cur: + cur.execute(select_item_location_sql, payload) + rows = cur.fetchone() + if rows and convert: + item_locations = postsqldb.tupleDictionaryFactory(cur.description, rows) + elif rows and not convert: + item_locations = rows + return item_locations + except Exception as error: + return error + +def selectCostLayersTuple(site_name, payload, convert=True): + """select a single or series of cost layers from the database for site_name + + Args: + conn (_T_connector@connect): + site_name (str): + payload (tuple): (item_locations_id, ) + convert (bool): defaults to False, used for determining return as tuple/dict + + Returns: + list: list of tuples/dict from the cost_layers table for site_name + """ + cost_layers = () + database_config = config.config() + select_cost_layers_sql = f"SELECT cl.* FROM {site_name}_item_locations il JOIN {site_name}_cost_layers cl ON cl.id = ANY(il.cost_layers) where il.id=%s;" + try: + with psycopg2.connect(**database_config) as conn: + with conn.cursor() as cur: + cur.execute(select_cost_layers_sql, payload) + rows = cur.fetchall() + if rows and convert: + cost_layers = rows + cost_layers = [postsqldb.tupleDictionaryFactory(cur.description, layer) for layer in rows] + elif rows and not convert: + cost_layers = rows + return cost_layers + except Exception as error: + return error + +def selectSiteTuple(payload, convert=True): + """Select a single Site from sites using site_name + + Args: + conn (_T_connector@connect): Postgresql Connector + payload (tuple): (site_name,) + convert (bool, optional): determines if to return tuple as dictionary. Defaults to False. + + Raises: + DatabaseError: + + Returns: + tuple or dict: selected tuples + """ + site = () + database_config = config.config() + select_site_sql = f"SELECT * FROM sites WHERE site_name = %s;" + try: + with psycopg2.connect(**database_config) as conn: + with conn.cursor() as cur: + cur.execute(select_site_sql, payload) + rows = cur.fetchone() + if rows and convert: + site = postsqldb.tupleDictionaryFactory(cur.description, rows) + elif rows and not convert: + site = rows + except Exception as error: + raise postsqldb.DatabaseError(error, payload, select_site_sql) + return site + def paginateZonesBySku(site: str, payload: tuple, convert=True): database_config = config.config() zones, count = (), 0 @@ -726,92 +823,48 @@ def insertItemTuple(site, payload, convert=True, conn=None): except Exception as error: raise postsqldb.DatabaseError(error, payload, sql) -def selectItemLocationsTuple(site_name, payload, convert=True): - """select a single tuple from ItemLocations table for site_name +def insertSKUPrefixtuple(site, payload, convert=True, conn=None): + """insert payload into zones table of site - Args: - conn (_T_connector@connect): - site_name (str): - payload (tuple): [item_id, location_id] - convert (bool): defaults to False, used to determine return of tuple/dict + Args: + conn (_T_connector@connect): Postgresql Connector + site (str): + payload (tuple): (name[str],) + convert (bool, optional): Determines if to return tuple as dictionary. Defaults to False. + + Raises: + DatabaseError: + + Returns: + tuple or dict: inserted tuple + """ + prefix = () + self_conn = False + with open(f"application/items/sql/insertSKUPrefixTuple.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 - Returns: - tuple: the row that was returned from the table - """ - item_locations = () - database_config = config.config() - select_item_location_sql = f"SELECT * FROM {site_name}_item_locations WHERE part_id = %s AND location_id = %s;" - try: - with psycopg2.connect(**database_config) as conn: with conn.cursor() as cur: - cur.execute(select_item_location_sql, payload) + cur.execute(sql, payload) rows = cur.fetchone() if rows and convert: - item_locations = postsqldb.tupleDictionaryFactory(cur.description, rows) + prefix = postsqldb.tupleDictionaryFactory(cur.description, rows) elif rows and not convert: - item_locations = rows - return item_locations - except Exception as error: - return error + prefix = rows -def selectCostLayersTuple(site_name, payload, convert=True): - """select a single or series of cost layers from the database for site_name + if self_conn: + conn.commit() + conn.close() - Args: - conn (_T_connector@connect): - site_name (str): - payload (tuple): (item_locations_id, ) - convert (bool): defaults to False, used for determining return as tuple/dict - - Returns: - list: list of tuples/dict from the cost_layers table for site_name - """ - cost_layers = () - database_config = config.config() - select_cost_layers_sql = f"SELECT cl.* FROM {site_name}_item_locations il JOIN {site_name}_cost_layers cl ON cl.id = ANY(il.cost_layers) where il.id=%s;" - try: - with psycopg2.connect(**database_config) as conn: - with conn.cursor() as cur: - cur.execute(select_cost_layers_sql, payload) - rows = cur.fetchall() - if rows and convert: - cost_layers = rows - cost_layers = [postsqldb.tupleDictionaryFactory(cur.description, layer) for layer in rows] - elif rows and not convert: - cost_layers = rows - return cost_layers - except Exception as error: - return error - -def selectSiteTuple(payload, convert=True): - """Select a single Site from sites using site_name - - Args: - conn (_T_connector@connect): Postgresql Connector - payload (tuple): (site_name,) - convert (bool, optional): determines if to return tuple as dictionary. Defaults to False. - - Raises: - DatabaseError: - - Returns: - tuple or dict: selected tuples - """ - site = () - database_config = config.config() - select_site_sql = f"SELECT * FROM sites WHERE site_name = %s;" - try: - with psycopg2.connect(**database_config) as conn: - with conn.cursor() as cur: - cur.execute(select_site_sql, payload) - rows = cur.fetchone() - if rows and convert: - site = postsqldb.tupleDictionaryFactory(cur.description, rows) - elif rows and not convert: - site = rows - except Exception as error: - raise postsqldb.DatabaseError(error, payload, select_site_sql) - return site + return prefix + except Exception as error: + raise postsqldb.DatabaseError(error, payload, sql) + def postDeleteCostLayer(site_name, payload, convert=True, conn=None): """ diff --git a/application/items/items_API.py b/application/items/items_API.py index 29969ff..3b6008d 100644 --- a/application/items/items_API.py +++ b/application/items/items_API.py @@ -21,6 +21,7 @@ from user_api import login_required import application.postsqldb as db from application.items import database_items from application.items import items_processes +import application.database_payloads as dbPayloads items_api = Blueprint('items_api', __name__, template_folder="templates", static_folder="static") @@ -645,20 +646,42 @@ def addBlankItem(): @items_api.route('/addSKUPrefix', methods=["POST"]) def addSKUPrefix(): + """ POST new SKU Prefix to the system given a uuid, name, description + --- + parameters: + - in: query + name: uuid + schema: + type: string + default: 1 + required: true + description: uuid for the sku which will be attached to items + - in: query + name: name + schema: + type: string + default: 1 + required: true + description: name of the Prefix + - in: query + name: description + schema: + type: string + default: 1 + required: true + description: description of the Prefix. + responses: + 200: + description: Prefix added successfully. + """ if request.method == "POST": - database_config = config() site_name = session['selected_site'] - try: - with psycopg2.connect(**database_config) as conn: - prefix = db.SKUPrefixTable.Payload( - request.get_json()['uuid'], - request.get_json()['name'], - request.get_json()['description'] - ) - db.SKUPrefixTable.insert_tuple(conn, site_name, prefix.payload()) - except Exception as error: - conn.rollback() - return jsonify({'error': True, 'message': error}) + prefix = dbPayloads.SKUPrefixPayload( + request.get_json()['uuid'], + request.get_json()['name'], + request.get_json()['description'] + ) + database_items.insertSKUPrefixtuple(site_name, prefix.payload()) return jsonify({'error': False, 'message': 'Prefix added!!'}) return jsonify({'error': True, 'message': 'These was an error with adding this Prefix!'}) diff --git a/application/items/items_processes.py b/application/items/items_processes.py index 453bbba..1d8d28e 100644 --- a/application/items/items_processes.py +++ b/application/items/items_processes.py @@ -1,11 +1,20 @@ +# 3rd party imports +import datetime +import psycopg2 +import json + +# applications imports from application.items import database_items import application.postsqldb as db import application.database_payloads as dbPayloads import config -import datetime -import psycopg2 -import json +""" + +items_processes.py handles more higher order workflows that a single database call would not be able to accomplish +or when more complex logics are needed. + +""" def postNewBlankItem(site_name: str, user_id: int, data: dict, conn=None): """ data = {'barcode', 'name', 'subtype'}""" diff --git a/application/items/sql/insertSKUPrefixTuple.sql b/application/items/sql/insertSKUPrefixTuple.sql new file mode 100644 index 0000000..555146f --- /dev/null +++ b/application/items/sql/insertSKUPrefixTuple.sql @@ -0,0 +1,4 @@ +INSERT INTO %%site_name%%_sku_prefix +(uuid, name, description) +VALUES (%s, %s, %s) +RETURNING *; \ No newline at end of file