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

fix(api): resolve SQL unpacking crash and Event serialization errors

Browse Source 
- Refactor SQL helpers in lib_sql_search to return empty tuples instead of False
- Add Pydantic pre-validators to Event_Base to coerce time objects to strings
- Improves API stability for Event searches and filtered lists
This commit is contained in:
Scott Idem
2026-01-20 15:49:13 -05:00
parent dc7732ab5f
commit e16fbaa34b
2 changed files with 30 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
app/lib_sql_search.py
View File

@@ -8,15 +8,15 @@ from sqlalchemy import text
log = logging.getLogger(__name__)
def sql_limit_offset_part(limit: int, offset: int = 0) -> str|bool:
def sql_limit_offset_part(limit: int, offset: int = 0) -> str:
"""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
return ''
def sql_and_like_part(and_like_dict_obj: dict) -> tuple[str, dict]|bool:
def sql_and_like_part(and_like_dict_obj: dict) -> tuple[str, dict]:
"""Creates a partial SQL string for AND LIKE queries."""
data = {}
if and_like_dict_obj and isinstance(and_like_dict_obj, dict):
@@ -26,9 +26,9 @@ def sql_and_like_part(and_like_dict_obj: dict) -> tuple[str, dict]|bool:
clauses.append(f"{key} LIKE :and_like_{key}")
data[f'and_like_{key}'] = value
return f"AND ({' AND '.join(clauses)})", data
return False
return '', {}
def sql_or_like_part(or_like_dict_obj: dict) -> tuple[str, dict]|bool:
def sql_or_like_part(or_like_dict_obj: dict) -> tuple[str, dict]:
"""Creates a partial SQL string for OR LIKE queries."""
data = {}
if or_like_dict_obj and isinstance(or_like_dict_obj, dict):
@@ -38,9 +38,9 @@ def sql_or_like_part(or_like_dict_obj: dict) -> tuple[str, dict]|bool:
clauses.append(f"{key} LIKE :or_like_{key}")
data[f'or_like_{key}'] = value
return f"AND ({' OR '.join(clauses)})", data
return False
return '', {}
def sql_and_in_dict_li_part(and_in_dict_li_dict_obj: dict) -> tuple[str, dict]|bool:
def sql_and_in_dict_li_part(and_in_dict_li_dict_obj: dict) -> tuple[str, dict]:
"""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):
@@ -50,9 +50,9 @@ def sql_and_in_dict_li_part(and_in_dict_li_dict_obj: dict) -> tuple[str, dict]|b
clauses.append(f"{key} IN :and_in_{key}")
data[f'and_in_{key}'] = value
return f"AND ({' AND '.join(clauses)})", data
return False
return '', {}
def sql_and_qry_part(and_qry_dict_obj: dict) -> tuple[str, dict]|bool:
def sql_and_qry_part(and_qry_dict_obj: dict) -> tuple[str, dict]:
"""Creates a partial SQL string for additional AND queries (equals)."""
data = {}
if and_qry_dict_obj and isinstance(and_qry_dict_obj, dict):
@@ -62,9 +62,9 @@ def sql_and_qry_part(and_qry_dict_obj: dict) -> tuple[str, dict]|bool:
clauses.append(f"{key} = :and_{key}")
data[f'and_{key}'] = value
return f"AND ({' AND '.join(clauses)})", data
return False
return '', {}
def sql_fulltext_qry_part(fulltext_qry_dict: dict) -> tuple[str, dict]|bool:
def sql_fulltext_qry_part(fulltext_qry_dict: dict) -> tuple[str, dict]:
"""Creates a partial SQL string for fulltext search."""
data = {}
if fulltext_qry_dict and isinstance(fulltext_qry_dict, dict):
@@ -74,12 +74,12 @@ 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
return '', {}
def sql_enable_part(table_name: str, enabled: str) -> tuple[str, bool|None]|bool:
def sql_enable_part(table_name: str, enabled: str) -> tuple[str, bool|None]:
"""Handles enabled/disabled status filtering with schema check."""
from app import lib_sql_core
if not table_name: return False
if not table_name: return '', None
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'all': return '', None
try:
@@ -89,12 +89,12 @@ def sql_enable_part(table_name: str, enabled: str) -> tuple[str, bool|None]|bool
return '', None
val = (enabled == 'enabled')
return f"AND `{table_name}`.enable = {str(val).lower()}", val
return False
return '', None
def sql_hidden_part(table_name: str, hidden: str) -> tuple[str, bool|None]|bool:
def sql_hidden_part(table_name: str, hidden: str) -> tuple[str, bool|None]:
"""Handles hidden status filtering with schema check."""
from app import lib_sql_core
if not table_name: return False
if not table_name: return '', None
if hidden in ['hidden', 'not_hidden', 'all']:
if hidden == 'all': return '', None
try:
@@ -105,9 +105,9 @@ def sql_hidden_part(table_name: str, hidden: str) -> tuple[str, bool|None]|bool:
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
return '', None
def sql_where_qry_part(qry_dict_li: list) -> tuple[str, dict]|bool:
def sql_where_qry_part(qry_dict_li: list) -> tuple[str, dict]:
"""Standard v2 style WHERE clause builder."""
data = {}
if qry_dict_li and isinstance(qry_dict_li, list):
@@ -124,7 +124,7 @@ def sql_where_qry_part(qry_dict_li: list) -> tuple[str, dict]|bool:
clauses.append(f'{type_} {field} {op} :{field}')
data[field] = val
return ' '.join(clauses), data
return False
return '', {}
def sql_search_qry_part(
search_query: Any,

14
app/models/event_models.py
View File

@@ -70,8 +70,8 @@ class Event_Base(BaseModel):
recurring: Optional[bool]
recurring_pattern: Optional[str]
recurring_start_time: Optional[datetime.time]
recurring_end_time: Optional[datetime.time]
recurring_start_time: Optional[str]
recurring_end_time: Optional[str]
recurring_text: Optional[str]
weekday_sunday: Optional[bool]
@@ -281,6 +281,12 @@ class Event_Base(BaseModel):
return v.astimezone(pytz.UTC).isoformat()
else: return v
@validator('recurring_start_time', 'recurring_end_time', pre=True, always=True)
def time_to_str(cls, v):
if isinstance(v, (datetime.time, datetime.timedelta)):
return str(v)
return v
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
@@ -340,8 +346,8 @@ class Event_Meeting_Flat_Base(BaseModel):
recurring: Optional[bool]
recurring_pattern: Optional[str]
recurring_start_time: Optional[datetime.time]
recurring_end_time: Optional[datetime.time]
recurring_start_time: Optional[str]
recurring_end_time: Optional[str]
recurring_text: Optional[str]
weekday_sunday: Optional[bool]