from __future__ import annotations import datetime from typing import Dict, List, Optional, Set, Union from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator from app.db_sql import redis_lookup_id_random, sql_delete, sql_enable_part, sql_insert, sql_insert_or_update, sql_limit_offset_part, sql_select, sql_update from app.lib_general import log, logging, logger_reset from app.methods.order_cfg_methods import load_order_cfg_obj from app.methods.order_line_methods import check_order_obj_line_list, create_order_obj_line, get_order_line_rec_list, load_order_obj_line, remove_order_obj_line, update_order_obj_line # from app.methods.person_methods import load_person_obj from app.models.common_field_schema import default_num_bytes # from app.models.order_models import Order_Base # from app.models.order_line_models import Order_Line_Base, Order_Line_DB_Base # This should go away later. from app.models.order_models_v3 import Order_Base from app.models.order_line_models_v3 import Order_Line_Base, Order_Line_DB_Base # This should go away later. # from app.models.person_models import Person_Base # ### BEGIN ### API Order Methods ### create_order_obj() ### # Updated 2022-01-18 def create_order_obj( account_id: int, order_dict_obj: Order_Base, person_id: int|None = None, ) -> int|bool: log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) # ### SECTION ### Secondary data validation log.info('Create dictionary or Pydantic object') log.debug(type(order_dict_obj)) if isinstance(order_dict_obj, dict): order_dict = order_dict_obj try: order_obj = Order_Base(**order_dict) except ValidationError as e: log.error(e.json()) return False else: order_obj = order_dict_obj # order_obj.account_id = account_id order_dict = order_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'cfg', 'order_line_list', 'person', 'user', 'created_on', 'updated_on'}) log.debug(order_obj) # ### SECTION ### Process data # Look for an account_id in the order_obj # if account_id: pass # elif account_id := order_obj.account_id: pass order_obj.account_id = account_id # Is this needed? order_dict['account_id'] = account_id # Look for a person_id in the contact_obj if person_id: order_obj.person_id = person_id # Is this needed? order_dict['person_id'] = person_id elif person_id := order_obj.person.id: pass if order_dict_in_result := sql_insert( data = order_dict, table_name = 'order', rm_id_random = True, id_random_length = default_num_bytes ): pass else: log.warning(f'Order not created.') log.debug(order_dict_in_result) return False log.debug(order_dict_in_result) order_id = order_dict_in_result if order_obj.order_line_list: log.info('Looping through Order Line list') for order_line in order_obj.order_line_list: # NOTE: Should there be a check here using check_order_obj_line_list()??? # In theory there should not already be any order_line records associated with the new order_id. order_line_id = order_line.id if order_line_id: log.info('Updating Order Line') if update_order_obj_result := update_order_obj_line( order_line_id = order_line_id, order_line_dict_obj = order_line, ): pass else: return False else: log.info('Creating Order Line') if create_order_obj_line_result := create_order_obj_line( order_id = order_id, order_line_dict_obj = order_line, ): pass else: pass log.info(f'Returning the Order ID: {order_id}') return order_id # ### END ### API Order Methods ### create_order_obj() ### # ### BEGIN ### API Order Methods ### update_order_obj() ### # Updated 2022-01-18 def update_order_obj( order_id: int, order_dict_obj: Order_Base, # person_id: int|None = None, ) -> int|bool: log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) # ### SECTION ### Secondary data validation log.info('Create dictionary or Pydantic object') log.debug(type(order_dict_obj)) if isinstance(order_dict_obj, dict): order_dict = order_dict_obj try: order_obj = Order_Base(**order_dict) except ValidationError as e: log.error(e.json()) return False else: order_obj = order_dict_obj # order_obj.account_id = account_id order_dict = order_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'cfg', 'order_line_list', 'person', 'user', 'created_on', 'updated_on'}) log.debug(order_obj) # ### SECTION ### Process data order_obj.id = order_id # Is this needed? order_dict['id'] = order_id # # Look for a person_id in the order_obj # if person_id: # order_obj.person_id = person_id # Is this needed? # order_dict['person_id'] = person_id # elif person_id := order_obj.person.id: pass if order_dict_up_result := sql_update( data = order_dict, table_name = 'order', rm_id_random = True, ): pass elif order_dict_up_result is None: pass # More than likely no order specific data was passed. There might be order lines though. else: log.warning(f'Person not updated.') log.debug(order_dict_up_result) return False log.debug(order_dict_up_result) if order_obj.order_line_list: log.info('Looping through Order Line list') for order_line in order_obj.order_line_list: order_line_id = order_line.id product_id = order_line.product_id if order_line.remove_line: remove_order_obj_line( order_id = order_id, order_line_id = order_line_id, product_id = product_id, ) continue # Do not need to check for anything else on this loop order_line_id = order_line.id product_id = order_line.product_id if for_person_id := order_line.for_person_id: pass log.info(f'Checking for matching order lines. Order ID: {order_id}; Product ID: {product_id}; For Person ID: {for_person_id}') order_line_id_found = None if check_order_obj_line_list_result := check_order_obj_line_list( order_id = order_id, product_id = product_id, for_person_id = for_person_id, ): for order_line_rec in check_order_obj_line_list_result: order_line_id_found = order_line_rec.get('order_line_id') if order_line_id == order_line_id_found: pass elif order_line_id := order_line_id_found: pass if order_line_id: log.info('Updating Order Line') if update_order_obj_result := update_order_obj_line( order_line_id = order_line_id, order_line_dict_obj = order_line, ): pass else: return False else: log.info('Creating Order Line') if create_order_obj_line_result := create_order_obj_line( order_id = order_id, order_line_dict_obj = order_line, ): pass else: pass return True # ### END ### API Order Methods ### update_order_obj() ### # ### BEGIN ### API Order Methods ### load_order_obj() ### # Updated 2021-11-19 # @logger_reset def load_order_obj( order_id: int|str, inc_address: bool = False, inc_contact: bool = False, inc_order_cfg: bool = False, inc_order_line_list: bool = False, inc_person: bool = False, inc_user: bool = False, enabled: str = 'enabled', # enabled, disabled, all limit: int = 500, offset: int = 0, by_alias: bool = True, exclude_unset: bool = True, model_as_dict: bool = False, ) -> None|bool|dict|list: log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) if order_id := redis_lookup_id_random(record_id_random=order_id, table_name='order'): pass else: return False # None, false bool if order_rec := sql_select(table_name='v_order', record_id=order_id): pass else: return order_rec # None, empty dict, empty list, false bool log.debug(order_rec) try: order_obj = Order_Base(**order_rec) except ValidationError as e: log.error(e.json()) return False log.debug(order_obj) # Updated 2022-01-18 if inc_order_cfg: # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL if order_cfg_result := load_order_cfg_obj( account_id = order_rec.get('account_id'), by_alias = by_alias, exclude_unset = exclude_unset, model_as_dict = True, ): order_obj.cfg = order_cfg_result else: order_obj.cfg = {} # None log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(order_cfg_result) # Updated 2021-06-18 if inc_order_line_list: if order_line_rec_list_result := get_order_line_rec_list( for_obj_type = 'order', for_obj_id = order_id, # product_for_type = 'all', # status = 'all', limit = limit, ): order_line_result_list = [] for order_line_rec in order_line_rec_list_result: order_line_result_list.append( load_order_obj_line( order_line_id = order_line_rec.get('order_line_id'), by_alias = by_alias, exclude_unset = exclude_unset, model_as_dict = model_as_dict, ) ) order_obj.order_line_list = order_line_result_list else: order_obj.order_line_list = [] # None # Updated 2022-01-18 if inc_person: from app.methods.person_methods import load_person_obj if person_result := load_person_obj( person_id = order_rec.get('person_id'), inc_address = inc_address, inc_contact = inc_contact, inc_user = inc_user, # NOTE: enabled = enabled, limit = limit, offset = offset, by_alias = by_alias, exclude_unset = exclude_unset, model_as_dict = model_as_dict, ): order_obj.person = person_result else: order_obj.person = {} # None pass # Updated 2021-06-22 # NOTE: Phasing out! Use *inc_user* under load_person_obj() instead. # if inc_user: # log.warning(f'This is being deprecated? load_order_obj() inc_user') # from app.methods.user_methods import load_user_obj # if user_result := load_user_obj( # user_id = order_rec.get('user_id', None), # limit = limit, # by_alias = by_alias, # exclude_unset = exclude_unset, # model_as_dict = model_as_dict, # enabled = enabled, # ): # order_obj.user = user_result # else: order_obj.user = None # pass if model_as_dict: return order_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member else: return order_obj # ### END ### API Order Methods ### load_order_obj() ### # ### BEGIN ### API Order Methods ### get_order_obj_cart_for_person_id_v3() ### # Updated 2022-01-21 @logger_reset def get_order_id_cart_for_person_id_v3( person_id: int|str, enabled: str = 'enabled', # enabled, disabled, all limit: int = 5, offset: int = 0, by_alias: bool = True, exclude_unset: bool = True, model_as_dict: bool = False, ) -> None|bool|dict|list: log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass else: return False # None, false bool data = {} data['person_id'] = person_id sql_enabled, data['enable'] = sql_enable_part(table_name='order', enabled=enabled) # Reasonably safe return str and bool sql_limit = sql_limit_offset_part(limit=limit, offset=offset) # Reasonably safe return str sql = f""" SELECT `order`.id AS 'order_id', `order`.id_random AS 'order_id_random' FROM `order` AS `order` WHERE `order`.person_id = :person_id AND `order`.status IN ('open', 'locked') {sql_enabled} ORDER BY `order`.created_on DESC, `order`.updated_on DESC {sql_limit}; """ log.debug(sql) if order_rec_result := sql_select(data=data, sql=sql): log.debug(order_rec_result) if isinstance(order_rec_result, dict): order_id_cart = order_rec_result.get('order_id') log.info(f'Got Order ID {order_id_cart} cart for Person ID {person_id}') return order_id_cart elif isinstance(order_rec_result, list): log.warning(f'Got multiple Orders for a cart for Person ID {person_id}. This should not happen.') return False else: # None or [] or False log.debug(order_rec_result) return order_rec_result # ### END ### API Order Methods ### get_order_obj_cart_for_person_id_v3() ### # ### BEGIN ### API Order Methods ### save_order_obj() ### # @logger_reset def save_order_obj(order_obj_new:Order_Base, repl_order_line_li: bool=False): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) log.debug(order_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True)) order_line_obj_li_curr = [] # Initialize to store order_line list if order_obj_new.id_random: log.info(f'An order.id {order_obj_new.id} or order.id_random {order_obj_new.id_random} was included. We can update an existing order.') log.info(f'Get the current order_line list to compare with what was sent...') data = {} data['order_id_random'] = order_obj_new.id_random if order_line_rec_li_curr := sql_select(table_name='v_order_line', data=data, rm_id_random=True, as_list=True): #log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(order_line_rec_li_curr) for order_line_rec in order_line_rec_li_curr: try: order_line_obj = Order_Line_Base(**order_line_rec) log.debug(order_line_obj) except ValidationError as e: log.error(e.json()) order_line_obj_li_curr.append(order_line_obj) else: log.info(f'No order_line records were found') elif order_obj_new.account_id_random and (order_obj_new.person_id_random or order_obj_new.user_id_random): log.info(f'An account.id_random {order_obj_new.account_id_random} was passed. And either a person.id_random {order_obj_new.person_id_random} or user.id_random {order_obj_new.user_id_random} was passed. We can create a new order.') # Because there was not an order ID, assume there are no order lines yet. So no look up. else: log.info('Either an order ID is required to update an order or an account ID along with a person ID or user ID is required to create an order.') return False #log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(order_line_obj_li_curr) if repl_order_line_li: # This will remove any order line list items not sent with the new order information. log.info('Removing any order line list items not sent with the new order information...') for index, order_line_obj_curr in enumerate(order_line_obj_li_curr): log.info(f'Current: order line ID={order_line_obj_curr.id_random} and product ID= {order_line_obj_curr.product_id_random}') matched_product_id = False for order_line_obj_new in order_obj_new.order_line_li: log.debug(f'Checking new: product ID={order_line_obj_new.product_id_random}') if order_line_obj_curr.product_id_random == order_line_obj_new.product_id_random: matched_product_id = True log.debug(f'Matched: product ID={order_line_obj_new.product_id_random}') break else: log.debug(f'No match: product ID={order_line_obj_new.product_id_random}') if not matched_product_id: # Was not found in the new order line list sent log.info(f'Current order line product ID did not match any of the new list. DELETE order line ID {order_line_obj_curr.id_random} with product ID {order_line_obj_curr.product_id_random}') if order_line_del_result := sql_delete(table_name='order_line', record_id_random=order_line_obj_curr.id_random): log.info(f'Deleted record and now pop the current list item {index}...') order_line_obj_li_curr.pop(index) else: log.info(f'Current order line product ID matched. Keeping order line ID {order_line_obj_curr.id_random} with product ID {order_line_obj_curr.product_id_random}') # NOTE: That this current order line item will be updated below. log.debug(order_line_obj_li_curr) #log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.info('Loop through the line list that was sent and compare with what was pulled from the DB') # Loop through the new line list that was sent and compare with the current line list that was pulled from the DB # Only insert if a product ID does not match # Only update if a product ID does match for order_line_obj_new in order_obj_new.order_line_li: log.info(f'New: order line ID={order_line_obj_new.id_random} and product ID= {order_line_obj_new.product_id_random}') matched_product_id = False for index, order_line_obj_curr in enumerate(order_line_obj_li_curr): log.debug(f'Checking current: product ID={order_line_obj_curr.product_id_random}') if order_line_obj_new.product_id_random == order_line_obj_curr.product_id_random: matched_product_id = True log.debug(f'Matched: product ID={order_line_obj_curr.product_id_random}') log.info(f'Updating the current line item with the new line item.') order_line_obj_new.id_random = order_line_obj_curr.id_random order_line_obj_new.id = order_line_obj_curr.id order_line_obj_li_curr[index] = order_line_obj_new break else: log.debug(f'No match: product ID={order_line_obj_curr.product_id_random}') if not matched_product_id: # Was not found in the current order line list that was pulled from the DB log.info(f'New order line product ID did not match any of the current list. Append order line ID {order_line_obj_new.id_random} with product ID {order_line_obj_new.product_id_random}') log.info('Append to current list...') order_line_obj_li_curr.append(order_line_obj_new) # These will be inserted/updated below # Save merged current and new list to the new order object order_obj_new.order_line_li = order_line_obj_li_curr log.debug(order_obj_new) # Final loop through to get the new order totals # Calculate totals order_total_amount: int = 0 order_total_quantity: int = 0 for order_line_obj_new in order_obj_new.order_line_li: order_total_amount += order_line_obj_new.quantity * order_line_obj_new.amount order_total_quantity += order_line_obj_new.quantity order_obj_new.total_bill = order_total_amount # "amount" is used by order_cart and Stripe order_obj_new.total_quantity = order_total_quantity # order_obj_new.balance = order_total_amount - order_obj_new.total_paid # No longer used 2022-03-16 log.debug(order_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'order_line_li', 'cfg', 'created_on', 'updated_on'})) order_obj_data = order_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'order_line_li', 'cfg', 'created_on', 'updated_on'}) # SQL INSERT or UPDATE the order record log.info('SQL INSERT or UPDATE the order record') if order_obj_resp := sql_insert_or_update(data=order_obj_data, table_name='order', rm_id_random=True, id_random_length=8): pass else: return False #log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(order_obj_resp) if isinstance(order_obj_resp, bool) and order_obj_resp: if order_id := order_obj_new.id: pass elif order_id := order_obj_new.id_random: pass elif isinstance(order_obj_resp, int): order_id = order_obj_resp else: return False log.debug(f'Order ID={order_id}') # Loop through the order_line list to SQL INSERT or UPDATE the records log.info('Loop through the order_line list to SQL INSERT or UPDATE the records') for order_line_obj_new in order_obj_new.order_line_li: log.info(f"New order_line: order_line_id_random={order_line_obj_new.id_random}; product_id_random={order_line_obj_new.product_id_random}") log.debug(order_line_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=False, exclude={'order_line_id_random', 'product_type_id', 'product_type', 'created_on', 'updated_on'})) order_line_obj_data = order_line_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'order_line_id_random', 'product_type_id', 'product_type', 'created_on', 'updated_on'}) # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL try: order_line_obj_db = Order_Line_DB_Base(**order_line_obj_new) log.debug(order_line_obj_db) except ValidationError as e: log.error(e.json()) return False order_line_obj_db_data = order_line_obj_db.dict(by_alias=False, exclude_defaults=False, exclude_unset=True) log.debug(order_line_obj_db_data) order_line_obj_data['order_id'] = order_id if order_line_obj_resp := sql_insert_or_update(sql=None, data=order_line_obj_data, table_name='order_line', rm_id_random=True, id_random_length=8): pass else: return False log.debug(order_line_obj_resp) return order_id # ### END ### API Order Methods ### save_order_obj() ### # ### BEGIN ### API Order Methods ### get_order_rec_list() ### # Updated 2021-12-13 # @logger_reset def get_order_rec_list( for_obj_type: str, for_obj_id: str, from_datetime: datetime.datetime = None, to_datetime: datetime.datetime = None, # balance_gt: int = 0, # $0 to $99999 status: list|str = 'closed', # started, in progress, complete, all # checkout_status: str = 'none', # none, canceled, waiting, success, failed, unknown enabled: str = 'enabled', # enabled, disabled, all limit: int = 500, offset: int = 0, ) -> list|bool: log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) # if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name=for_obj_type): pass # else: return False data = {} if for_obj_type == 'account' and for_obj_id: data['account_id'] = redis_lookup_id_random(record_id_random=for_obj_id, table_name='account') sql_obj_type_id = f'`order`.account_id = :account_id' elif for_obj_type == 'person' and for_obj_id: data['person_id'] = redis_lookup_id_random(record_id_random=for_obj_id, table_name='person') sql_obj_type_id = f'`order`.person_id = :person_id' else: return False # allowed_status_li = ['started', 'in progress', 'complete', 'all'] # OLD list # allowed_status_li = ['open', 'locked', 'reopened', 'closed', 'canceled', 'other'] # NEW list allowed_status_li = ['open', 'locked', 'in progress', 'reopened', 'closed', 'complete', 'canceled', 'other'] # TEMPORARY list... sql_status = '' if isinstance(status, list): data['status'] = status sql_status = f'AND `order`.status IN :status' elif status in allowed_status_li: if status == 'closed' or status == 'complete': data['status'] = ['closed', 'complete'] sql_status = f'AND `order`.status IN :status' elif status == 'locked' or status == 'in progress': data['status'] = ['locked', 'in progress'] sql_status = f'AND `order`.status IN :status' else: data['status'] = status sql_status = f'AND `order`.status = :status' elif status == 'all': sql_status = f'AND `order`.status IS NOT NULL' else: log.warning('The status value passed is not allowed. Returning None') return False if from_datetime and to_datetime: data['from_datetime'] = from_datetime data['to_datetime'] = to_datetime sql_from_to_datetime = f'AND `order`.created_on >= :from_datetime AND `order`.created_on <= :to_datetime' elif from_datetime: data['from_datetime'] = from_datetime sql_from_to_datetime = f'AND `order`.created_on >= :from_datetime' elif to_datetime: data['to_datetime'] = to_datetime sql_from_to_datetime = f'AND `order`.created_on <= :to_datetime' else: sql_from_to_datetime = '' sql_enabled, data['enable'] = sql_enable_part(table_name='order', enabled=enabled) # Reasonably safe return str and bool sql_limit = sql_limit_offset_part(limit=limit, offset=offset) # Reasonably safe return str log.debug(data) sql = f""" SELECT `order`.id AS 'order_id', `order`.id_random AS 'order_id_random' FROM `order` AS `order` WHERE {sql_obj_type_id} {sql_status} {sql_from_to_datetime} {sql_enabled} ORDER BY `order`.created_on DESC, `order`.updated_on DESC {sql_limit}; """ log.debug(sql) if order_rec_li_result := sql_select(data=data, sql=sql, as_list=True): log.info('Got a list result') order_rec_li = order_rec_li_result else: # [] or False log.info('No results or something went wrong') order_rec_li = order_rec_li_result log.debug(order_rec_li_result) return order_rec_li # ### END ### API Order Methods ### get_order_rec_list() ###