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

Updated API CRUD and SQL SELECT related functions. They can now handle multiple ANDs!

Browse Source
This commit is contained in:
Scott Idem
2023-11-30 19:59:38 -05:00
parent 6d1cc6c1ff
commit 4b6c048eaf
2 changed files with 258 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

270
app/db_sql.py
View File

@@ -541,6 +541,8 @@ def sql_select(
field_value = None,
enabled: str|None = None, # enabled, disabled, all
hidden: str|None = None, # hidden, not_hidden, all
fulltext_qry_dict: dict|None = None,
and_qry_dict: dict|None = None,
fulltext_qry_field_li: list|None = None, # ['field_name_1', 'field_name_2']
fulltext_qry_str: str|None = None, # 'search string'
order_by_li: dict|None = None, # {"the_field_name": "DESC"}
@@ -555,6 +557,7 @@ def sql_select(
log_lvl: int = logging.WARNING, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
) -> None|bool|dict|list:
log.setLevel(log_lvl)
log.debug(locals())
if limit >= 0 and offset >= 0:
log.info(f'Creating partial SQL string for LIMIT and OFFSET. Limit: {limit}; Offset: {offset}')
@@ -596,21 +599,20 @@ def sql_select(
# NOTE: Version 2 of the fulltext search
# NOTE: This version works well and can do multiple MATCH AGAINST at a time. - STI 2023-11-29
sql_fulltext_match_against = ''
log.debug(fulltext_qry_field_li)
if fulltext_qry_field_li and isinstance(fulltext_qry_field_li, list) and fulltext_qry_str: # fulltext_qry_field_li should be a list
log.info('Creating partial SQL string for fulltext search.')
fulltext_qry_field_li_str = []
for value in fulltext_qry_field_li:
log.debug(value)
fulltext_qry_field_li_str.append(f'MATCH( {value} ) AGAINST( :fulltext_qry_str IN BOOLEAN MODE )')
fulltext_qry_field_string = ' OR '.join(fulltext_qry_field_li_str)
# sql_fulltext_match_against = ''
# log.debug(fulltext_qry_field_li)
# if fulltext_qry_field_li and isinstance(fulltext_qry_field_li, list) and fulltext_qry_str: # fulltext_qry_field_li should be a list
# log.info('Creating partial SQL string for fulltext search.')
# fulltext_qry_field_li_str = []
# for value in fulltext_qry_field_li:
# log.debug(value)
# fulltext_qry_field_li_str.append(f'MATCH( {value} ) AGAINST( :fulltext_qry_str IN BOOLEAN MODE )')
# fulltext_qry_field_string = ' OR '.join(fulltext_qry_field_li_str)
sql_fulltext_match_against = f'AND ({fulltext_qry_field_string})'
log.debug(sql_fulltext_match_against)
# sql_fulltext_match_against = f'AND ({fulltext_qry_field_string})'
# log.debug(sql_fulltext_match_against)
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# if enabled:
# sql_enabled = sql_enable_part(table_name=table_name, enabled=enabled) # Reasonably safe return str
@@ -690,16 +692,57 @@ def sql_select(
)
elif table_name and field_name and field_value and not (record_id or record_id_random or sql or data):
# Select all records from a table with a specific field and field value
# Updated 2023-07-06
# Updated 2023-11-30
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.info('Select all records from a table with a specific field and field value')
if not data:
data = {}
sql_fulltext_match_against = ''
if fulltext_qry_dict:
sql_fulltext_match_against, data_qry = sql_fulltext_qry_part(fulltext_qry_dict)
# NOTE: Merge the data_qry result with the data dict
data = {**data, **data_qry}
sql_and_qry = ''
if and_qry_dict:
sql_and_qry, data_qry = sql_and_qry_part(and_qry_dict)
# NOTE: Merge the data_qry result with the data dict
data = {**data, **data_qry}
# # NOTE: Version 3 of the fulltext search
# sql_fulltext_match_against = ''
# log.debug(fulltext_qry_dict)
# if fulltext_qry_dict and isinstance(fulltext_qry_dict, dict): # fulltext_qry_dict should be a dict
# log.info('Creating partial SQL string for fulltext search.')
# fulltext_qry_dict_str = []
# # if not data:
# # data = {}
# for key, value in fulltext_qry_dict.items():
# log.debug(f'Key = {key}; Value = {value}')
# fulltext_qry_dict_str.append(f'MATCH( {key} ) AGAINST( :ft_{key} IN BOOLEAN MODE )')
# # fulltext_qry_dict_str.append(f'MATCH( {key} ) AGAINST( :{key} IN BOOLEAN MODE )')
# data[f'ft_{key}'] = value
# # data[key] = 'temp value'
# # log.debug(data)
# # data[key] = value
# log.debug(data)
# fulltext_qry_field_string = ' OR '.join(fulltext_qry_dict_str)
# sql_fulltext_match_against = f'AND ({fulltext_qry_field_string})'
# log.debug(sql_fulltext_match_against)
# NOTE: This is new and currently only working with the API CRUD list endpoint and the sql_select function calls. -2023-07-06
# NOTE: This may need more testing.
data = {}
# if not data:
# data = {}
data[field_name] = field_value
if sql_fulltext_match_against:
data['fulltext_qry_str'] = fulltext_qry_str
# if sql_fulltext_match_against:
# data['fulltext_qry_str'] = fulltext_qry_str
if enabled:
sql_enabled, data['enabled'] = sql_enable_part(table_name=table_name, enabled=enabled) # Reasonably safe return str
@@ -729,6 +772,7 @@ def sql_select(
FROM `{table_name}`
WHERE `{table_name}`.{field_name} = :{field_name}
{sql_fulltext_match_against}
{sql_and_qry}
{sql_enabled}
{sql_hidden}
{sql_order_by}
@@ -780,6 +824,75 @@ def sql_select(
log.warning('Nothing matched the expected combination of parameters passed to this function')
return False # Not successful
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(sql)
log.debug(data)
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
result = run_sql_select(sql=sql, data=data)
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(f'Row count: {result.rowcount}')
# log.debug(vars(result))
# log.debug(dir(result))
# NOTE: as_dict defaults to True for this function
# NOTE: as_list defaults to False for this function
# NOTE: After testing, this method is the fastest way to convert to a dict - STI 2021-03-09
# NOTE: My custom sql_result_proxy_to_dict_simple(result_proxy=result.first()) is slower than using dict().
# NOTE: list(result) was tested seems to be the slowest. Slower than my custom function.
if result.rowcount == 1:
log.info(f'Found one record. as_dict={as_dict}, as_list={as_list}')
if as_dict:
record = dict(result.first())
else:
record = result.first()
if as_list:
record_li = []
record_li.append(record)
log.debug(record_li)
return record_li # Successful
else:
log.debug(record)
return record # Successful
elif result.rowcount > 1:
log.info(f'Found {result.rowcount} records. as_dict={as_dict}, as_list={as_list}')
if as_dict:
#timer_1_start = timer()
record_li = [dict(record) for record in result.fetchall()]
#log.debug(record_li)
#log.debug(type(record_li))
#log.debug(type(record_li[0]))
#timer_1_end = timer()
#log.debug( round((timer_1_end - timer_1_start), 8) )
else:
record_li = result.fetchall()
log.debug(record_li)
return record_li # Successful
else:
if as_list:
log.info('No records found. Returning an empty list.')
log.debug(result)
return [] # Successful even though no results
else:
log.info('No records found. Returning None.')
log.debug(result)
return None # Successful even though no results
# ### END ### Core Help CRUD ### sql_select() ###
# ### BEGIN ### Core Help CRUD ### run_sql_select() ###
# Updated 2023-11-29
@logger_reset
def run_sql_select(
sql: str|None = None,
data: dict|None = None,
log_lvl: int = logging.WARNING, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
) -> None|bool|dict|list:
log.setLevel(log_lvl)
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug('*** ** * ** ***')
log.debug(sql)
@@ -857,59 +970,8 @@ def sql_select(
else:
log.debug('Successfully executed the SQL on the first try.')
pass
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(f'Row count: {result.rowcount}')
# log.debug(vars(result))
# log.debug(dir(result))
# NOTE: as_dict defaults to True for this function
# NOTE: as_list defaults to False for this function
# NOTE: After testing, this method is the fastest way to convert to a dict - STI 2021-03-09
# NOTE: My custom sql_result_proxy_to_dict_simple(result_proxy=result.first()) is slower than using dict().
# NOTE: list(result) was tested seems to be the slowest. Slower than my custom function.
if result.rowcount == 1:
log.info(f'Found one record. as_dict={as_dict}, as_list={as_list}')
if as_dict:
record = dict(result.first())
else:
record = result.first()
if as_list:
record_li = []
record_li.append(record)
log.debug(record_li)
return record_li # Successful
else:
log.debug(record)
return record # Successful
elif result.rowcount > 1:
log.info(f'Found {result.rowcount} records. as_dict={as_dict}, as_list={as_list}')
if as_dict:
#timer_1_start = timer()
record_li = [dict(record) for record in result.fetchall()]
#log.debug(record_li)
#log.debug(type(record_li))
#log.debug(type(record_li[0]))
#timer_1_end = timer()
#log.debug( round((timer_1_end - timer_1_start), 8) )
else:
record_li = result.fetchall()
log.debug(record_li)
return record_li # Successful
else:
if as_list:
log.info('No records found. Returning an empty list.')
log.debug(result)
return [] # Successful even though no results
else:
log.info('No records found. Returning None.')
log.debug(result)
return None # Successful even though no results
# ### END ### Core Help CRUD ### sql_select() ###
return result
# ### END ### Core Help CRUD ### run_sql_select() ###
# ### BEGIN ### Core Help CRUD ### sql_delete() ###
# The catch all SQL DELETE function - STI 2021-02-17
@@ -1094,8 +1156,8 @@ def redis_lookup_id_random(
record_id_random: int|str,
table_name: str,
check_int_id: bool = False,
log_lvl: int = logging.INFO, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
minutes: int = 7,
log_lvl: int = logging.WARNING, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
minutes: int = 1,
) -> str|int|bool|None:
log.setLevel(log_lvl)
@@ -1523,6 +1585,70 @@ def get_account_id_w_for_type_id(
# ### END ### API DB SQL Methods ### get_account_id_w_for_type_id() ###
# ### BEGIN ### API DB SQL Methods ### sql_fulltext_qry_part() ###
# Updated 2023-11-30
@logger_reset
def sql_fulltext_qry_part(
fulltext_qry_dict: dict, # One or more key value pairs. key = field name; value = search string
) -> bool|dict:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# NOTE: Version 3 of the fulltext search
data = {}
sql_fulltext_match_against = ''
log.debug(fulltext_qry_dict)
if fulltext_qry_dict and isinstance(fulltext_qry_dict, dict):
log.info('Creating partial SQL string for fulltext search.')
fulltext_qry_dict_str = []
for key, value in fulltext_qry_dict.items():
log.debug(f'Key = {key}; Value = {value}')
fulltext_qry_dict_str.append(f'MATCH( {key} ) AGAINST( :ft_{key} IN BOOLEAN MODE )')
data[f'ft_{key}'] = value
fulltext_qry_field_string = ' OR '.join(fulltext_qry_dict_str)
sql_fulltext_match_against = f'AND ({fulltext_qry_field_string})'
log.debug(sql_fulltext_match_against)
log.debug(data)
return sql_fulltext_match_against, data
# ### BEGIN ### API DB SQL Methods ### sql_and_qry_part() ###
# Updated 2023-11-30
@logger_reset
def sql_and_qry_part(
and_qry_dict_obj: dict, # One or more key value pairs. key = field name; value = search string
) -> bool|dict:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# NOTE: Version 3 of the fulltext search
data = {}
sql_and_qry = ''
log.debug(and_qry_dict_obj)
if and_qry_dict_obj and isinstance(and_qry_dict_obj, dict):
log.info('Creating partial SQL string for additional AND queries.')
and_qry_dict_obj_str = []
for key, value in and_qry_dict_obj.items():
log.debug(f'Key = {key}; Value = {value}')
and_qry_dict_obj_str.append(f'{key} = :and_{key}')
data[f'and_{key}'] = value
and_qry_field_string = ' AND '.join(and_qry_dict_obj_str)
sql_and_qry = f'AND ({and_qry_field_string})'
log.debug(sql_and_qry)
log.debug(data)
return sql_and_qry, data
# ### BEGIN ### API DB SQL Methods ### sql_enable_part() ###
# Updated 2022-01-17
@logger_reset