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

refactor(sql): modularize basic search query builders

Browse Source
This commit is contained in:
Scott Idem
2026-01-06 17:22:10 -05:00
parent d584457997
commit b5e874bd99
2 changed files with 89 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
app/db_sql.py
View File

@@ -11,6 +11,15 @@ from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError
from sqlalchemy.pool import NullPool
# from multiprocessing import Pool
from app.lib_sql_search import (
sql_limit_offset_part as _sql_limit_offset_part,
sql_and_like_part as _sql_and_like_part,
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
)
db_uri = settings.SQLALCHEMY_DB_URI
@@ -1876,30 +1885,7 @@ def sql_fulltext_qry_part(
) -> bool|dict:
log.setLevel(logging.INFO) # 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 )')
# fulltext_qry_dict_str.append(f'MATCH( {key} ) AGAINST( :ft_{key} IN NATURAL LANGUAGE 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
return _sql_fulltext_qry_part(fulltext_qry_dict)
# ### BEGIN ### API DB SQL Methods ### sql_and_qry_part() ###
@@ -1910,28 +1896,7 @@ def sql_and_qry_part(
) -> 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
return _sql_and_qry_part(and_qry_dict_obj)
# ### BEGIN ### API DB SQL Methods ### sql_and_like_part() ###
@@ -1942,28 +1907,7 @@ def sql_and_like_part(
) -> bool|dict:
log.setLevel(logging.INFO)
log.debug(locals())
data = {}
sql_and_like = ''
log.debug(and_like_dict_obj)
if and_like_dict_obj and isinstance(and_like_dict_obj, dict):
log.info('Creating partial SQL string for additional AND LIKE queries.')
and_like_dict_obj_str = []
for key, value in and_like_dict_obj.items():
log.debug(f'Key = {key}; Value = {value}')
and_like_dict_obj_str.append(f'{key} LIKE :and_like_{key}')
# For now not surrounding with %... may need to be added back in later
# data[f'and_like_{key}'] = f'%{value}%'
data[f'and_like_{key}'] = f'{value}'
and_like_field_string = ' AND '.join(and_like_dict_obj_str)
sql_and_like = f'AND ({and_like_field_string})'
log.debug(sql_and_like)
log.debug(data)
return sql_and_like, data
return _sql_and_like_part(and_like_dict_obj)
# ### BEGIN ### API DB SQL Methods ### sql_or_like_part() ###
@@ -1974,28 +1918,7 @@ def sql_or_like_part(
) -> bool|dict:
log.setLevel(logging.INFO)
log.debug(locals())
data = {}
sql_or_like = ''
log.debug(or_like_dict_obj)
if or_like_dict_obj and isinstance(or_like_dict_obj, dict):
log.info('Creating partial SQL string for additional OR LIKE queries.')
or_like_dict_obj_str = []
for key, value in or_like_dict_obj.items():
log.debug(f'Key = {key}; Value = {value}')
or_like_dict_obj_str.append(f'{key} LIKE :or_like_{key}')
# For now not surrounding with %... may need to be added back in later
# data[f'or_like_{key}'] = f'%{value}%'
data[f'or_like_{key}'] = f'{value}'
or_like_field_string = ' OR '.join(or_like_dict_obj_str)
sql_or_like = f'AND ({or_like_field_string})'
log.debug(sql_or_like)
log.debug(data)
return sql_or_like, data
return _sql_or_like_part(or_like_dict_obj)
# ### END ### API DB SQL Methods ### sql_or_like_part() ###
@@ -2009,27 +1932,7 @@ def sql_and_in_dict_li_part(
) -> bool|dict:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = {}
sql_and_in_dict_li = ''
log.debug(and_in_dict_li_dict_obj)
if and_in_dict_li_dict_obj and isinstance(and_in_dict_li_dict_obj, dict):
log.info('Creating partial SQL string for additional AND IN queries.')
and_in_dict_li_dict_obj_str = []
for key, value in and_in_dict_li_dict_obj.items():
log.debug(f'Key = {key}; Value = {value}')
and_in_dict_li_dict_obj_str.append(f'{key} IN :and_in_{key}')
# and_in_dict_li_dict_obj_str.append(f'{key} IN ( :and_in_{key} )')
data[f'and_in_{key}'] = value
and_in_dict_li_field_string = ' AND '.join(and_in_dict_li_dict_obj_str)
sql_and_in_dict_li = f'AND ({and_in_dict_li_field_string})'
log.debug(sql_and_in_dict_li)
log.debug(data)
return sql_and_in_dict_li, data
return _sql_and_in_dict_li_part(and_in_dict_li_dict_obj)
# ### BEGIN ### API DB SQL Methods ### sql_enable_part() ###
@@ -2112,13 +2015,7 @@ def sql_hidden_part(table_name: str, hidden: str) -> bool|dict:
def sql_limit_offset_part(limit: int, offset: int = 0) -> bool|str:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if limit >= 0 and offset >= 0:
log.info(f'Creating partial SQL string for LIMIT and OFFSET. Limit: {limit}; Offset: {offset}')
sql = f'LIMIT {limit} OFFSET {offset}'
return sql
else:
return False
return _sql_limit_offset_part(limit, offset)
# ### END ### API DB SQL Methods ### sql_limit_offset_part() ###

74
app/lib_sql_search.py Normal file
View File

@@ -0,0 +1,74 @@
"""
Modular search builder and query generators for Aether.
"""
import logging
log = logging.getLogger(__name__)
def sql_limit_offset_part(limit: int, offset: int = 0) -> str|bool:
"""Creates a partial SQL string for LIMIT and OFFSET."""
if limit >= 0 and offset >= 0:
log.info(f'Creating partial SQL string for LIMIT and OFFSET. Limit: {limit}; Offset: {offset}')
return f'LIMIT {limit} OFFSET {offset}'
else:
return False
def sql_and_like_part(and_like_dict_obj: dict) -> tuple[str, dict]|bool:
"""Creates a partial SQL string for AND LIKE queries."""
data = {}
if and_like_dict_obj and isinstance(and_like_dict_obj, dict):
log.info('Creating partial SQL string for additional AND LIKE queries.')
clauses = []
for key, value in and_like_dict_obj.items():
clauses.append(f"{key} LIKE :and_like_{key}")
data[f'and_like_{key}'] = value
return f"AND ({' AND '.join(clauses)})", data
return False
def sql_or_like_part(or_like_dict_obj: dict) -> tuple[str, dict]|bool:
"""Creates a partial SQL string for OR LIKE queries."""
data = {}
if or_like_dict_obj and isinstance(or_like_dict_obj, dict):
log.info('Creating partial SQL string for additional OR LIKE queries.')
clauses = []
for key, value in or_like_dict_obj.items():
clauses.append(f"{key} LIKE :or_like_{key}")
data[f'or_like_{key}'] = value
return f"AND ({' OR '.join(clauses)})", data
return False
def sql_and_in_dict_li_part(and_in_dict_li_dict_obj: dict) -> tuple[str, dict]|bool:
"""Creates a partial SQL string for AND IN queries."""
data = {}
if and_in_dict_li_dict_obj and isinstance(and_in_dict_li_dict_obj, dict):
log.info('Creating partial SQL string for additional AND IN queries.')
clauses = []
for key, value in and_in_dict_li_dict_obj.items():
clauses.append(f"{key} IN :and_in_{key}")
data[f'and_in_{key}'] = value
return f"AND ({' AND '.join(clauses)})", data
return False
def sql_and_qry_part(and_qry_dict_obj: dict) -> tuple[str, dict]|bool:
"""Creates a partial SQL string for additional AND queries (equals)."""
data = {}
if and_qry_dict_obj and isinstance(and_qry_dict_obj, dict):
log.info('Creating partial SQL string for additional AND queries.')
clauses = []
for key, value in and_qry_dict_obj.items():
clauses.append(f"{key} = :and_{key}")
data[f'and_{key}'] = value
return f"AND ({' AND '.join(clauses)})", data
return False
def sql_fulltext_qry_part(fulltext_qry_dict: dict) -> tuple[str, dict]|bool:
"""Creates a partial SQL string for fulltext search."""
data = {}
if fulltext_qry_dict and isinstance(fulltext_qry_dict, dict):
log.info('Creating partial SQL string for fulltext search.')
clauses = []
for key, value in fulltext_qry_dict.items():
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