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

Less debug. Also why was this using the print() function? It should have been using the normal log.info() or whatever.

Browse Source
This commit is contained in:
Scott Idem
2026-01-20 19:27:16 -05:00
parent 45ca81a3e3
commit b2ee1f2760
2 changed files with 30 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
app/lib_sql_crud.py
View File

@@ -173,7 +173,7 @@ def sql_insert_or_update(
fields = [f'`{k}`' for k in data.keys() if k != 'id'] fields = [f'`{k}`' for k in data.keys() if k != 'id']
placeholders = [f':{k}' for k in data.keys() if k != 'id'] placeholders = [f':{k}' for k in data.keys() if k != 'id']
updates = [f'`{k}` = :{k}' for k in data.keys() if k != 'id'] updates = [f'`{k}` = :{k}' for k in data.keys() if k != 'id']
for k, v in data.items(): for k, v in data.items():
if isinstance(v, (dict, list)): if isinstance(v, (dict, list)):
data[k] = json.dumps(v) data[k] = json.dumps(v)
@@ -225,15 +225,15 @@ def sql_select(
log_lvl: int = logging.WARNING, log_lvl: int = logging.WARNING,
) -> None|bool|dict|list: ) -> None|bool|dict|list:
from app.lib_sql_search import ( from app.lib_sql_search import (
sql_enable_part, sql_hidden_part, sql_search_qry_part, sql_enable_part, sql_hidden_part, sql_search_qry_part,
sql_where_qry_part, sql_fulltext_qry_part, sql_and_qry_part, sql_where_qry_part, sql_fulltext_qry_part, sql_and_qry_part,
sql_and_like_part, sql_or_like_part, sql_and_in_dict_li_part sql_and_like_part, sql_or_like_part, sql_and_in_dict_li_part
) )
log.setLevel(log_lvl) log.setLevel(log_lvl)
sql_limit_offset = f'LIMIT {limit} OFFSET {offset}' if limit >= 0 and offset >= 0 else '' sql_limit_offset = f'LIMIT {limit} OFFSET {offset}' if limit >= 0 and offset >= 0 else ''
sql_order_by = '' sql_order_by = ''
if order_by_li and isinstance(order_by_li, dict): if order_by_li and isinstance(order_by_li, dict):
order_by_str_li = [f'`{table_name}`.`{k}` {v}' for k, v in order_by_li.items()] order_by_str_li = [f'`{table_name}`.`{k}` {v}' for k, v in order_by_li.items()]
@@ -245,19 +245,19 @@ def sql_select(
s_hi, d_hi = sql_hidden_part(table_name, hidden) if hidden else ('', None) s_hi, d_hi = sql_hidden_part(table_name, hidden) if hidden else ('', None)
if d_en is not None: data['enabled'] = d_en if d_en is not None: data['enabled'] = d_en
if d_hi is not None: data['hidden'] = d_hi if d_hi is not None: data['hidden'] = d_hi
s_search, d_search = ('', {}) s_search, d_search = ('', {})
if search_query: if search_query:
s_search, d_search = sql_search_qry_part(search_query, searchable_fields, table_name=table_name) s_search, d_search = sql_search_qry_part(search_query, searchable_fields, table_name=table_name)
data.update(d_search) data.update(d_search)
stmt = text(f"SELECT * FROM `{table_name}` WHERE 1=1 {s_search} {s_en} {s_hi} {sql_order_by} {sql_limit_offset};") stmt = text(f"SELECT * FROM `{table_name}` WHERE 1=1 {s_search} {s_en} {s_hi} {sql_order_by} {sql_limit_offset};")
elif table_name and (record_id or record_id_random) and not (field_name or field_value or sql or data): elif table_name and (record_id or record_id_random) and not (field_name or field_value or sql or data):
data = {'rid': record_id} if record_id else {'ridr': record_id_random} data = {'rid': record_id} if record_id else {'ridr': record_id_random}
where = f"`{table_name}`.id = :rid" if record_id else f"`{table_name}`.id_random = :ridr" where = f"`{table_name}`.id = :rid" if record_id else f"`{table_name}`.id_random = :ridr"
stmt = text(f"SELECT * FROM `{table_name}` WHERE {where} {sql_order_by} {sql_limit_offset};") stmt = text(f"SELECT * FROM `{table_name}` WHERE {where} {sql_order_by} {sql_limit_offset};")
elif table_name and field_name and field_value and not (record_id or record_id_random or sql or data): elif table_name and field_name and field_value and not (record_id or record_id_random or sql or data):
data = {field_name: field_value} data = {field_name: field_value}
s_where, d_where = sql_where_qry_part(qry_dict_li) if qry_dict_li else ('', {}) s_where, d_where = sql_where_qry_part(qry_dict_li) if qry_dict_li else ('', {})
@@ -269,7 +269,7 @@ def sql_select(
s_search, d_search = sql_search_qry_part(search_query, searchable_fields, table_name=table_name) if search_query else ('', {}) s_search, d_search = sql_search_qry_part(search_query, searchable_fields, table_name=table_name) if search_query else ('', {})
s_en, d_en = sql_enable_part(table_name, enabled) if enabled else ('', None) s_en, d_en = sql_enable_part(table_name, enabled) if enabled else ('', None)
s_hi, d_hi = sql_hidden_part(table_name, hidden) if hidden else ('', None) s_hi, d_hi = sql_hidden_part(table_name, hidden) if hidden else ('', None)
data.update(d_where); data.update(d_ft); data.update(d_and); data.update(d_alike) data.update(d_where); data.update(d_ft); data.update(d_and); data.update(d_alike)
data.update(d_olike); data.update(d_in); data.update(d_search) data.update(d_olike); data.update(d_in); data.update(d_search)
if d_en is not None: data['enabled'] = d_en if d_en is not None: data['enabled'] = d_en
@@ -296,11 +296,11 @@ def sql_select(
if count == 0: if count == 0:
return [] if as_list else None return [] if as_list else None
if count == 1: if count == 1:
record = dict(rows[0]) if as_dict else rows[0] record = dict(rows[0]) if as_dict else rows[0]
return [record] if as_list else record return [record] if as_list else record
# count > 1 # count > 1
records = [dict(r) for r in rows] if as_dict else rows records = [dict(r) for r in rows] if as_dict else rows
return records return records
@@ -315,9 +315,9 @@ def run_sql_select(
log_lvl: int = logging.WARNING, log_lvl: int = logging.WARNING,
) -> Any: ) -> Any:
log.setLevel(log_lvl) log.setLevel(log_lvl)
print(f"Executing SQL: {sql} with data: {data}", flush=True) # print(f"Executing SQL: {sql} with data: {data}", flush=True)
try: try:
return lib_sql_core.db.execute(sql, data) return lib_sql_core.db.execute(sql, data)
except (OperationalError, ProgrammingError) as e: except (OperationalError, ProgrammingError) as e:

32
app/lib_sql_search.py
View File

@@ -135,7 +135,7 @@ def sql_search_qry_part(
"""Recursively builds a SQL WHERE clause from a SearchQuery model.""" """Recursively builds a SQL WHERE clause from a SearchQuery model."""
from app import lib_sql_core from app import lib_sql_core
data = {} data = {}
param_counter = [0] param_counter = [0]
def get_param_name(): def get_param_name():
param_counter[0] += 1 param_counter[0] += 1
@@ -157,11 +157,11 @@ def sql_search_qry_part(
else: else:
use_match = True use_match = True
if table_name: if table_name:
try: try:
lib_sql_core.db.execute(text(f"SELECT default_qry_str FROM `{table_name}` LIMIT 0")) lib_sql_core.db.execute(text(f"SELECT default_qry_str FROM `{table_name}` LIMIT 0"))
except: except:
use_match = False use_match = False
else: else:
use_match = False use_match = False
if use_match: if use_match:
@@ -172,22 +172,22 @@ def sql_search_qry_part(
like_clauses = [] like_clauses = []
# Fields to exclude from a generic text 'q' search (numeric, technical, or date fields) # Fields to exclude from a generic text 'q' search (numeric, technical, or date fields)
exclude_patterns = [ exclude_patterns = [
'enable', 'hide', 'priority', 'sort', 'group', 'enable', 'hide', 'priority', 'sort', 'group',
'created_on', 'updated_on' 'created_on', 'updated_on'
] ]
for field in searchable_fields: for field in searchable_fields:
# Exclude internal integer IDs specifically # Exclude internal integer IDs specifically
if field.endswith('_id') or field == 'id': if field.endswith('_id') or field == 'id':
continue continue
# Exclude other technical/meta fields # Exclude other technical/meta fields
if any(x == field for x in exclude_patterns): if any(x == field for x in exclude_patterns):
continue continue
f_p_name = get_param_name() f_p_name = get_param_name()
like_clauses.append(f"`{field}` LIKE :{f_p_name}") like_clauses.append(f"`{field}` LIKE :{f_p_name}")
data[f_p_name] = f"%{query_node.query_string}%" data[f_p_name] = f"%{query_node.query_string}%"
if like_clauses: clauses.append(f"({' OR '.join(like_clauses)})") if like_clauses: clauses.append(f"({' OR '.join(like_clauses)})")
for filter_attr in ['and_filters', 'or_filters']: for filter_attr in ['and_filters', 'or_filters']:
if hasattr(query_node, filter_attr) and getattr(query_node, filter_attr): if hasattr(query_node, filter_attr) and getattr(query_node, filter_attr):
@@ -204,19 +204,19 @@ def sql_search_qry_part(
def process_filter(f) -> tuple[str, dict]: def process_filter(f) -> tuple[str, dict]:
# --- ID VISION MAPPING --- # --- ID VISION MAPPING ---
# If the frontend uses clean names (id, account_id), # If the frontend uses clean names (id, account_id),
# map them to the database columns (id_random, account_id_random) # map them to the database columns (id_random, account_id_random)
# ONLY if those columns actually exist in this table/view. # ONLY if those columns actually exist in this table/view.
target_field = f.field target_field = f.field
vision_fields = [ vision_fields = [
'id', 'account_id', 'site_id', 'person_id', 'user_id', 'id', 'account_id', 'site_id', 'person_id', 'user_id',
'journal_id', 'journal_entry_id', 'page_id', 'post_id', 'journal_id', 'journal_entry_id', 'page_id', 'post_id',
'post_comment_id', 'organization_id', 'address_id', 'hosted_file_id' 'post_comment_id', 'organization_id', 'address_id', 'hosted_file_id'
] ]
if target_field in vision_fields: if target_field in vision_fields:
candidate_field = 'id_random' if target_field == 'id' else f"{target_field}_random" candidate_field = 'id_random' if target_field == 'id' else f"{target_field}_random"
# Schema Check: Verify if the random version exists in the current table/view # Schema Check: Verify if the random version exists in the current table/view
use_random = False use_random = False
if table_name: if table_name:
@@ -225,10 +225,10 @@ def sql_search_qry_part(
use_random = True use_random = True
except Exception: except Exception:
pass pass
if use_random: if use_random:
target_field = candidate_field target_field = candidate_field
print(f"Search Trace: Mapping filter field '{f.field}' -> '{target_field}'", flush=True) # print(f"Search Trace: Mapping filter field '{f.field}' -> '{target_field}'", flush=True)
else: else:
# If random doesn't exist, we must stick to the integer column # If random doesn't exist, we must stick to the integer column
# but we'll need to resolve the string value to an integer elsewhere # but we'll need to resolve the string value to an integer elsewhere
@@ -239,7 +239,7 @@ def sql_search_qry_part(
# Fallback check for original field just in case # Fallback check for original field just in case
if f.field not in searchable_fields: if f.field not in searchable_fields:
raise HTTPException(status_code=400, detail=f"Unauthorized search field '{f.field}' (mapped to '{target_field}')") raise HTTPException(status_code=400, detail=f"Unauthorized search field '{f.field}' (mapped to '{target_field}')")
sql_op = operator_map.get(f.op.lower()) sql_op = operator_map.get(f.op.lower())
if not sql_op: raise HTTPException(status_code=400, detail=f"Unsupported operator: {f.op}") if not sql_op: raise HTTPException(status_code=400, detail=f"Unsupported operator: {f.op}")
filter_data = {} filter_data = {}