Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working on the cart and other related things

Browse Source
This commit is contained in:
Scott Idem
2021-05-10 18:29:53 -04:00
parent 8cbabf48fe
commit 9a46b755bf
36 changed files with 444 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
app/db_sql.py
View File

@@ -1,5 +1,5 @@
from __future__ import annotations
import secrets
import redis, secrets
from timeit import default_timer as timer
from app.config import settings
@@ -115,8 +115,8 @@ def sql_update(sql:str=None, data:dict=None, table_name:str=None, record_id:int=
if sql:
sql_update = text(sql)
elif table_name and data:
#if rm_id_random:
#data = lookup_id_random_pop(obj_data=data)
if rm_id_random:
data = lookup_id_random_pop(obj_data=data)
if not data.get('id_random', None) and id_random_length:
data['id_random'] = secrets.token_urlsafe(id_random_length)
@@ -200,6 +200,7 @@ def sql_insert_or_update(sql:str=None, data:dict=None, table_name:str=None, rm_i
elif table_name and data:
if rm_id_random:
data = lookup_id_random_pop(obj_data=data)
pass
if not data.get('id_random', None) and id_random_length:
data['id_random'] = secrets.token_urlsafe(id_random_length)
@@ -631,3 +632,210 @@ def sql_delete(
# NOTE: Need to deal with 0 rows affected when the WHERE clause was not satisfied and there was no error.
return True # Successful
# ### BEGIN ### API Lib General ### redis_lookup_id_random() ###
# Just return the value if it is an integer
# Check if the id_random value is a string and the correct length
# Attempt to look up id_random key in Redis
# If success then return the ID number
# If not success and there is a table_name then check the database table passed
# If found in database table then store in Redis and return the ID number
def redis_lookup_id_random(record_id_random:int|str, table_name:str):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if isinstance(record_id_random, str) and len(record_id_random) >= 11 and len(record_id_random) <= 22: pass
elif isinstance(record_id_random, int): return record_id_random
else:
log.warning(f'Unexpected data type or string format: {str(type(record_id_random))} Expected type is a string 11 or 22 characters long.')
return False
if record_id_random and table_name:
# WARNING: The record_id_random string length should be checked just in case?
if len(record_id_random) < 11:
log.warning(f'The length of id_random is too short: {str(record_id_random)} ({len(record_id_random)} chars)')
return False
elif len(record_id_random) > 22:
log.warning(f'The length of id_random is too long {str(record_id_random)} ({len(record_id_random)} chars)')
return False
else:
pass
elif record_id_random:
log.warning('Missing id_random')
return False
elif table_name:
log.warning('Missing table_name to select from for id_random')
return False
else:
log.warning('Missing table_name and record_id_random')
return False
r = redis.Redis(host='localhost', port=6379, db=7, password=None, decode_responses=True)
key_name = 'record_id:'+record_id_random
record_id = r.get(key_name)
log.debug(f'Record ID? {str(record_id)}')
if record_id:
log.info('The record ID was found using the record_id_random value.')
log.info(f'TTL for: {key_name} : {str(record_id)} is {str(r.ttl(key_name))} seconds')
return int(record_id)
elif table_name:
data = { 'id_random': record_id_random }
sql = f"""
SELECT id
FROM `{table_name}` AS `table`
WHERE `table`.id_random = :id_random;
"""
if select_results := sql_select(sql=sql, data=data):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(select_results)
log.debug(type(select_results))
if isinstance(select_results, dict):
log.info(f"""Record ID random found: {str(select_results['id'])}""")
if record_id := select_results.get('id'):
r.setex(key_name, datetime.timedelta(minutes=90), value=record_id)
return int(record_id)
else:
log.setLevel(logging.ERROR) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.error('The SQL result was not what was expected.')
return False
else:
log.setLevel(logging.ERROR) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.error('More than one record may have been found. There may be a duplicate id_random.')
log.error(select_results)
return False
else:
#log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.info('Record ID random was not found')
return None
log.setLevel(logging.ERROR) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.error('We should not be here. Something unexpected happened.')
return False # Just in case
# ### BEGIN ### API Lib General ### redis_lookup_id_random() ###
# ### BEGIN ### API Lib General ### lookup_id_random_pop() ###
# Look up and resolve id_random values to their id
# Remove the unneeded *_id_random key from the dict
def lookup_id_random_pop(obj_data:dict):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if 'account_id_random' in obj_data:
obj_data['account_id'] = redis_lookup_id_random(record_id_random=obj_data['account_id_random'], table_name='account')
obj_data.pop('account_id_random')
if 'address_id_random' in obj_data:
obj_data['address_id'] = redis_lookup_id_random(record_id_random=obj_data['address_id_random'], table_name='address')
obj_data.pop('address_id_random')
if 'address_location_id_random' in obj_data:
obj_data['address_location_id'] = redis_lookup_id_random(record_id_random=obj_data['address_location_id_random'], table_name='address')
obj_data.pop('address_location_id_random')
if 'archive_id_random' in obj_data:
obj_data['archive_id'] = redis_lookup_id_random(record_id_random=obj_data['archive_id_random'], table_name='archive')
obj_data.pop('archive_id_random')
if 'contact_id_random' in obj_data:
obj_data['contact_id'] = redis_lookup_id_random(record_id_random=obj_data['contact_id_random'], table_name='contact')
obj_data.pop('contact_id_random')
if 'event_id_random' in obj_data:
obj_data['event_id'] = redis_lookup_id_random(record_id_random=obj_data.get('event_id_random', None), table_name='event')
obj_data.pop('event_id_random')
if 'event_exhibit_id_random' in obj_data:
obj_data['event_exhibit_id'] = redis_lookup_id_random(record_id_random=obj_data.get('event_exhibit_id_random', None), table_name='event_exhibit')
obj_data.pop('event_exhibit_id_random')
if 'hosted_file_id_random' in obj_data:
obj_data['hosted_file_id'] = redis_lookup_id_random(record_id_random=obj_data.get('hosted_file_id_random', None), table_name='hosted_file')
obj_data.pop('hosted_file_id_random')
if 'order_id_random' in obj_data:
obj_data['order_id'] = redis_lookup_id_random(record_id_random=obj_data.get('order_id_random', None), table_name='order')
obj_data.pop('order_id_random')
if 'order_line_id_random' in obj_data:
obj_data['order_line_id'] = redis_lookup_id_random(record_id_random=obj_data.get('order_line_id_random', None), table_name='order_line')
obj_data.pop('order_line_id_random')
if 'order_cart_id_random' in obj_data:
obj_data['order_cart_id'] = redis_lookup_id_random(record_id_random=obj_data.get('order_cart_id_random', None), table_name='order_cart')
obj_data.pop('order_cart_id_random')
if 'order_cart_line_id_random' in obj_data:
obj_data['order_cart_line_id'] = redis_lookup_id_random(record_id_random=obj_data.get('order_cart_line_id_random', None), table_name='order_cart_line')
obj_data.pop('order_cart_line_id_random')
if 'organization_id_random' in obj_data:
obj_data['organization_id'] = redis_lookup_id_random(record_id_random=obj_data.get('organization_id_random', None), table_name='organization')
obj_data.pop('organization_id_random')
if 'person_id_random' in obj_data:
obj_data['person_id'] = redis_lookup_id_random(record_id_random=obj_data['person_id_random'], table_name='person')
obj_data.pop('person_id_random')
if 'post_id_random' in obj_data:
obj_data['post_id'] = redis_lookup_id_random(record_id_random=obj_data.get('post_id_random', None), table_name='post')
obj_data.pop('post_id_random')
if 'product_id_random' in obj_data:
obj_data['product_id'] = redis_lookup_id_random(record_id_random=obj_data['product_id_random'], table_name='product')
obj_data.pop('product_id_random')
if 'user_id_random' in obj_data:
obj_data['user_id'] = redis_lookup_id_random(record_id_random=obj_data['user_id_random'], table_name='user')
obj_data.pop('user_id_random')
if 'for_type' in obj_data and 'for_id_random' in obj_data:
obj_data['for_id'] = redis_lookup_id_random(record_id_random=obj_data.get('for_id_random', None), table_name=obj_data.get('for_type', None))
obj_data.pop('for_id_random')
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(obj_data)
elif 'for_id_random' in obj_data:
# In case for_id_random was passed without for_type
log.warn('for_id_random was passed without for_type')
obj_data.pop('for_id_random')
if 'object_type' in obj_data and 'object_id_random' in obj_data:
obj_data['object_id'] = redis_lookup_id_random(record_id_random=obj_data.get('object_id_random', None), table_name=obj_data.get('object_type', None))
obj_data.pop('object_id_random')
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(obj_data)
elif 'object_id_random' in obj_data:
# In case object_id_random was passed without object_type
log.warn('object_id_random was passed without object_type')
obj_data.pop('object_id_random')
if 'to_object_type' in obj_data and 'to_object_id_random' in obj_data:
obj_data['to_object_id'] = redis_lookup_id_random(record_id_random=obj_data.get('to_object_id_random', None), table_name=obj_data.get('to_object_type', None))
obj_data.pop('to_object_id_random')
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(obj_data)
elif 'to_object_id_random' in obj_data:
# In case to_object_id_random was passed without to_object_type
log.warn('to_object_id_random was passed without to_object_type')
obj_data.pop('to_object_id_random')
if 'from_object_type' in obj_data and 'from_object_id_random' in obj_data:
obj_data['from_object_id'] = redis_lookup_id_random(record_id_random=obj_data.get('from_object_id_random', None), table_name=obj_data.get('from_object_type', None))
obj_data.pop('from_object_id_random')
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(obj_data)
elif 'from_object_id_random' in obj_data:
# In case from_object_id_random was passed without from_object_type
log.warn('from_object_id_random was passed without from_object_type')
obj_data.pop('from_object_id_random')
return obj_data
# ### END ### API Lib General ### lookup_id_random_pop() ###