From a6a5162385c34318f9c82a05cd14b298371c4755 Mon Sep 17 00:00:00 2001 From: Scott Idem Date: Tue, 6 Jan 2026 17:24:47 -0500 Subject: [PATCH] refactor(sql): modularize status and where query builders --- app/db_sql.py | 130 +++--------------------------------------- app/lib_sql_search.py | 53 ++++++++++++++++- 2 files changed, 60 insertions(+), 123 deletions(-) diff --git a/app/db_sql.py b/app/db_sql.py index e14a4e0..4b9a09f 100644 --- a/app/db_sql.py +++ b/app/db_sql.py @@ -17,7 +17,10 @@ from app.lib_sql_search import ( sql_or_like_part as _sql_or_like_part, sql_and_in_dict_li_part as _sql_and_in_dict_li_part, sql_and_qry_part as _sql_and_qry_part, - sql_fulltext_qry_part as _sql_fulltext_qry_part + sql_fulltext_qry_part as _sql_fulltext_qry_part, + sql_enable_part as _sql_enable_part, + sql_hidden_part as _sql_hidden_part, + sql_where_qry_part as _sql_where_qry_part ) @@ -1802,78 +1805,13 @@ 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_where_qry_part() ### -# Example JSON data -# jp: { -# qry: [ -# { -# type: "AND", -# field: "enable", -# operator: "=", -# value: TRUE -# }, -# { -# type: "AND", -# field: "example", -# operator: ">=", -# value: 2 -# }, -# { -# type: "OR", -# field: "test", -# operator: "LIKE", -# value: "%xyz%" -# }, -# ] -# } -# Updated 2024-08-14 +@logger_reset def sql_where_qry_part( qry_dict_li: list, # JSON data ) -> bool|str: log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) - - data = {} - sql_where_qry = '' - - if qry_dict_li and isinstance(qry_dict_li, list): - log.info('Creating partial SQL string for WHERE queries.') - sql_where_qry_str = [] - - for qry in qry_dict_li: - log.debug(qry) - if qry.get('type') == '': - if qry.get("operator") == 'MATCH': - # sql_where_qry_str.append(f'{qry.get("field")} {qry.get("operator")} AGAINST(:{qry.get("field")} IN BOOLEAN MODE)') - sql_where_qry_str.append(f'AND MATCH( {qry.get("field")} ) AGAINST( :{qry.get("field")} IN BOOLEAN MODE )') - else: - sql_where_qry_str.append(f'AND {qry.get("field")} {qry.get("operator")} :{qry.get("field")}') - data[qry.get('field')] = qry.get('value') - elif qry.get('type') == 'AND': - if qry.get("operator") == 'MATCH': - # sql_where_qry_str.append(f'AND {qry.get("field")} {qry.get("operator")} AGAINST(:{qry.get("field")} IN BOOLEAN MODE)') - sql_where_qry_str.append(f'AND MATCH( {qry.get("field")} ) AGAINST( :{qry.get("field")} IN BOOLEAN MODE )') - else: - sql_where_qry_str.append(f'AND {qry.get("field")} {qry.get("operator")} :{qry.get("field")}') - data[qry.get('field')] = qry.get('value') - elif qry.get('type') == 'OR': - if qry.get("operator") == 'MATCH': - # sql_where_qry_str.append(f'OR {qry.get("field")} {qry.get("operator")} AGAINST(:{qry.get("field")} IN BOOLEAN MODE)') - sql_where_qry_str.append(f'OR MATCH( {qry.get("field")} ) AGAINST( :{qry.get("field")} IN BOOLEAN MODE )') - else: - sql_where_qry_str.append(f'OR {qry.get("field")} {qry.get("operator")} :{qry.get("field")}') - data[qry.get('field')] = qry.get('value') - else: - log.error(f'Unknown query type: {qry.get("type")}') - return False - - # Should this WHERE part also be surrounded by parentheses??? - # sql_where_qry = 'AND ('+' '.join(sql_where_qry_str)+')' - # sql_where_qry = sql_where_qry_str - sql_where_qry = ' '.join(sql_where_qry_str) - log.debug(sql_where_qry) - - return sql_where_qry, data + return _sql_where_qry_part(qry_dict_li) # ### END ### API DB SQL Methods ### sql_where_qry_part() ### @@ -1942,33 +1880,7 @@ def sql_and_in_dict_li_part( def sql_enable_part(table_name: str, enabled: str) -> bool|dict: log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) - - if not table_name: return False - - if enabled in ['enabled', 'disabled', 'all']: - log.info(f'Creating partial SQL string for "enabled" check. Enabled: {enabled}') - - if enabled == 'all': - return '', None - - # Check if column exists - try: - db.execute(text(f"SELECT enable FROM `{table_name}` LIMIT 0")) - except: - log.warning(f"Table/View '{table_name}' does not have an 'enable' column. Skipping filter.") - return '', None - - if enabled == 'enabled': - sql = f'AND `{table_name}`.enable = true' - enable = True - elif enabled == 'disabled': - sql = f'AND `{table_name}`.enable = false' - enable = False - - log.debug(sql) - return sql, enable - else: - return False + return _sql_enable_part(table_name, enabled) # ### END ### API DB SQL Methods ### sql_enable_part() ### @@ -1979,33 +1891,7 @@ def sql_enable_part(table_name: str, enabled: str) -> bool|dict: def sql_hidden_part(table_name: str, hidden: str) -> bool|dict: log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) - - if not table_name: return False - - if hidden in ['hidden', 'not_hidden', 'all']: - log.info(f'Creating partial SQL string for "hidden" check. Hide: {hidden}') - - if hidden == 'all': - return '', None - - # Check if column exists - try: - db.execute(text(f"SELECT hide FROM `{table_name}` LIMIT 0")) - except: - log.warning(f"Table/View '{table_name}' does not have a 'hide' column. Skipping filter.") - return '', None - - if hidden == 'hidden': - sql = f'AND `{table_name}`.hide = true' - hide = True - elif hidden == 'not_hidden': - sql = f'AND (`{table_name}`.hide = false OR `{table_name}`.hide IS NULL)' - hide = False - - log.debug(sql) - return sql, hide - else: - return False + return _sql_hidden_part(table_name, hidden) # ### END ### API DB SQL Methods ### sql_enable_part() ### diff --git a/app/lib_sql_search.py b/app/lib_sql_search.py index 56c1adb..d9e0cbc 100644 --- a/app/lib_sql_search.py +++ b/app/lib_sql_search.py @@ -2,6 +2,7 @@ Modular search builder and query generators for Aether. """ import logging +from sqlalchemy import text log = logging.getLogger(__name__) @@ -71,4 +72,54 @@ def sql_fulltext_qry_part(fulltext_qry_dict: dict) -> tuple[str, dict]|bool: clauses.append(f"MATCH( {key} ) AGAINST( :ft_{key} IN BOOLEAN MODE )") data[f'ft_{key}'] = value return f"AND ({' OR '.join(clauses)})", data - return False \ No newline at end of file + return False + +def sql_enable_part(table_name: str, enabled: str) -> tuple[str, bool|None]|bool: + """Handles enabled/disabled status filtering with schema check.""" + from app.db_sql import db + if not table_name: return False + if enabled in ['enabled', 'disabled', 'all']: + if enabled == 'all': return '', None + try: + db.execute(text(f"SELECT enable FROM `{table_name}` LIMIT 0")) + except: + log.warning(f"Table '{table_name}' missing 'enable' column. Skipping filter.") + return '', None + val = (enabled == 'enabled') + return f"AND `{table_name}`.enable = {str(val).lower()}", val + return False + +def sql_hidden_part(table_name: str, hidden: str) -> tuple[str, bool|None]|bool: + """Handles hidden status filtering with schema check.""" + from app.db_sql import db + if not table_name: return False + if hidden in ['hidden', 'not_hidden', 'all']: + if hidden == 'all': return '', None + try: + db.execute(text(f"SELECT hide FROM `{table_name}` LIMIT 0")) + except: + log.warning(f"Table '{table_name}' missing 'hide' column. Skipping filter.") + return '', None + if hidden == 'hidden': + return f"AND `{table_name}`.hide = true", True + return f"AND (`{table_name}`.hide = false OR `{table_name}`.hide IS NULL)", False + return False + +def sql_where_qry_part(qry_dict_li: list) -> tuple[str, dict]|bool: + """Standard v2 style WHERE clause builder.""" + data = {} + if qry_dict_li and isinstance(qry_dict_li, list): + log.info('Creating partial SQL string for WHERE queries.') + clauses = [] + for qry in qry_dict_li: + field = qry.get('field') + op = qry.get('operator') + val = qry.get('value') + type_ = qry.get('type', 'AND') or 'AND' + if op == 'MATCH': + clauses.append(f'{type_} MATCH( {field} ) AGAINST( :{field} IN BOOLEAN MODE )') + else: + clauses.append(f'{type_} {field} {op} :{field}') + data[field] = val + return ' '.join(clauses), data + return False