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

Feature: Implement advanced POST-based search with recursive logical grouping and unique parameterization (Verified Working).

Browse Source
This commit is contained in:
Scott Idem
2026-01-02 17:09:29 -05:00
parent 7b9ec69e7b
commit 2f24a5588b
4 changed files with 200 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
app/db_sql.py
View File

@@ -1,4 +1,5 @@
import datetime, json, pytz, random, redis, secrets
from typing import Any, Optional
from timeit import default_timer as timer
from app.config import settings
@@ -614,6 +615,7 @@ def sql_select(
and_like_dict: dict|None = None,
or_like_dict: dict|None = None,
and_in_dict_li: dict|None = None,
search_query: Any|None = None, # NEW 2026-01-02 (SearchQuery model)
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"}
@@ -715,17 +717,21 @@ def sql_select(
else:
sql_hidden = ''
# data['hidden'] = ''
# if sql_enabled:
# data['enable'] = sql_enable_part(table_name=table_name, enabled=enabled) # Reasonably safe return str and bool
# if sql_hidden:
# sql_hidden = sql_hidden_part(table_name=table_name, hidden=hidden) # Reasonably safe return str and bool
# sql_hidden, data['hidden'] = sql_hidden_part(table_name=table_name, hidden=hidden) # Reasonably safe return str and bool
sql_search_qry = ''
if search_query:
log.info('Creating partial SQL string for complex SearchQuery.')
sql_search_qry, data_search = sql_search_qry_part(search_query)
data = {**data, **data_search}
sql = text(
f"""
SELECT *
FROM `{table_name}`
WHERE 1=1
{sql_search_qry}
{sql_enabled}
{sql_hidden}
{sql_order_by}
{sql_limit_offset}
;
@@ -819,6 +825,12 @@ def sql_select(
# NOTE: Merge the data_qry result with the data dict
data = {**data, **data_qry}
sql_search_qry = ''
if search_query:
log.info('Creating partial SQL string for complex SearchQuery.')
sql_search_qry, data_search = sql_search_qry_part(search_query)
data = {**data, **data_search}
# # NOTE: Version 3 of the fulltext search
# sql_fulltext_match_against = ''
# log.debug(fulltext_qry_dict)
@@ -883,6 +895,7 @@ def sql_select(
{sql_and_like}
{sql_or_like}
{sql_and_in_dict_li}
{sql_search_qry}
{sql_enabled}
{sql_hidden}
{sql_order_by}
@@ -2089,3 +2102,96 @@ def sql_limit_offset_part(limit: int, offset: int = 0) -> bool|str:
else:
return False
# ### END ### API DB SQL Methods ### sql_limit_offset_part() ###
# ### BEGIN ### API DB SQL Methods ### sql_search_qry_part() ###
# NEW 2026-01-02
# Updated to support complex POST-based searches with recursive logical grouping.
@logger_reset
def sql_search_qry_part(
search_query: Any, # SearchQuery model instance
) -> tuple[str, dict]:
"""
Recursively builds a SQL WHERE clause from a SearchQuery model.
Uses unique parameter names to prevent collisions.
"""
log.setLevel(logging.INFO)
log.debug(locals())
data = {}
param_counter = [0] # List used as a reference for unique parameter names
def get_param_name():
param_counter[0] += 1
return f"sp_{param_counter[0]}"
operator_map = {
"eq": "=",
"ne": "!=",
"gt": ">",
"gte": ">=",
"lt": "<",
"lte": "<=",
"like": "LIKE",
"in": "IN",
"is_null": "IS NULL",
"is_not_null": "IS NOT NULL"
}
def process_node(query_node) -> str:
clauses = []
# Process 'and' filters
if hasattr(query_node, 'and_filters') and query_node.and_filters:
and_clauses = []
for item in query_node.and_filters:
if hasattr(item, 'field'): # SearchFilter
clause, item_data = process_filter(item)
and_clauses.append(clause)
data.update(item_data)
else: # Nested SearchQuery
and_clauses.append(f"({process_node(item)})")
if and_clauses:
clauses.append(f"({' AND '.join(and_clauses)})")
# Process 'or' filters
if hasattr(query_node, 'or_filters') and query_node.or_filters:
or_clauses = []
for item in query_node.or_filters:
if hasattr(item, 'field'): # SearchFilter
clause, item_data = process_filter(item)
or_clauses.append(clause)
data.update(item_data)
else: # Nested SearchQuery
or_clauses.append(f"({process_node(item)})")
if or_clauses:
clauses.append(f"({' OR '.join(or_clauses)})")
return ' AND '.join(clauses)
def process_filter(f) -> tuple[str, dict]:
sql_op = operator_map.get(f.op.lower())
if not sql_op:
raise ValueError(f"Unsupported search operator: {f.op}")
filter_data = {}
if f.op.lower() in ['is_null', 'is_not_null']:
clause = f"`{f.field}` {sql_op}"
elif f.op.lower() == 'in':
p_name = get_param_name()
# IN operator requires a tuple or list
clause = f"`{f.field}` IN (:{p_name})"
filter_data[p_name] = f.value
else:
p_name = get_param_name()
clause = f"`{f.field}` {sql_op} :{p_name}"
filter_data[p_name] = f.value
return clause, filter_data
# Initial processing
sql_where = process_node(search_query)
if sql_where:
return f"AND ({sql_where})", data
return "", {}
# ### END ### API DB SQL Methods ### sql_search_qry_part() ###

24
app/models/api_crud_models.py
View File

@@ -1,6 +1,6 @@
import datetime, pytz
from typing import Dict, List, Optional, Set, Union
from typing import Any, Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random
@@ -11,6 +11,28 @@ log = logging.getLogger(__name__)
from app.models.common_field_schema import base_fields, default_num_bytes
# ### BEGIN ### API Search Models ###
class SearchFilter(BaseModel):
"""
Represents a single filter condition.
Example: {"field": "price", "op": "gt", "value": 100}
"""
field: str
op: str # eq, ne, gt, gte, lt, lte, like, in, is_null, is_not_null
value: Optional[Any] = None
class SearchQuery(BaseModel):
"""
Represents a complex search query with optional logical grouping.
"""
and_filters: Optional[List[Union[SearchFilter, 'SearchQuery']]] = Field(None, alias="and")
or_filters: Optional[List[Union[SearchFilter, 'SearchQuery']]] = Field(None, alias="or")
# Support recursive models in Pydantic v1
SearchQuery.update_forward_refs()
# ### END ### API Search Models ###
# ### BEGIN ### API CRUD Models ### Fundraising_Cfg_Base() ###
class Api_Crud_Base(BaseModel):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL

62
app/routers/api_crud_v3.py
View File

@@ -16,6 +16,7 @@ from app.lib_general_v3 import (
DelayParams, get_delay_params
)
from app.models.response_models import *
from app.models.api_crud_models import SearchQuery
from app.ae_obj_types_def import obj_type_kv_li
from app.db_sql import redis_lookup_id_random, sql_select, sql_insert, sql_update, sql_delete, get_id_random
@@ -213,6 +214,67 @@ async def get_obj_li(
return mk_resp(data=[], status_code=200, response=response) # Return empty list on no results
@router.post('/{obj_type_l1}/search', response_model=Resp_Body_Base, tags=['CRUD v3 Search (Dev)'])
async def search_obj_li(
response: Response,
obj_type_l1: str,
search_query: SearchQuery,
order_by_li: Optional[str] = Query(None),
account: AccountContext = Depends(get_account_context),
pagination: PaginationParams = Depends(get_pagination_params),
status_filter: StatusFilterParams = Depends(get_status_filter_params),
serialization: SerializationParams = Depends(get_serialization_params),
delay: DelayParams = Depends(get_delay_params),
):
"""
Search top-level objects using a complex SearchQuery in the POST body.
This endpoint supports:
- Recursive AND/OR grouping
- Operators: eq, ne, gt, gte, lt, lte, like, in, is_null, is_not_null
- Large filters that would exceed URL length limits.
"""
if delay.sleep_time_s > 0:
await asyncio.sleep(delay.sleep_time_s)
log.setLevel(logging.WARNING)
log.debug(locals())
if order_by_li:
order_by_li = json.loads(order_by_li)
obj_name = obj_type_l1
if obj_name not in obj_type_kv_li:
return mk_resp(data=False, status_code=400, response=response, status_message=f"Object type '{obj_name}' not found.")
obj_cfg = obj_type_kv_li[obj_name]
table_name = obj_cfg.get('tbl_default', obj_cfg.get('tbl'))
base_name = obj_cfg.get('mdl_default', obj_cfg.get('mdl'))
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration for object type '{obj_name}' is incomplete.")
sql_result = sql_select(
table_name=table_name,
enabled=status_filter.enabled,
hidden=status_filter.hidden,
search_query=search_query,
order_by_li=order_by_li,
limit=pagination.limit,
offset=pagination.offset,
as_list=True,
)
if sql_result:
resp_data_li = []
for record in sql_result:
resp_data = base_name(**record).dict(by_alias=serialization.by_alias, exclude_unset=serialization.exclude_unset)
resp_data_li.append(resp_data)
return mk_resp(data=resp_data_li, response=response)
else:
return mk_resp(data=[], status_code=200, response=response)
@router.post('/{obj_type_l1}/', response_model=Resp_Body_Base)
async def post_obj(
request: Request,

4
documentation/V3_CRUD_ARCHITECTURE_AND_LEARNINGS.md
View File

@@ -16,6 +16,10 @@ The V3 CRUD API (`/v3/crud/`) is designed to run in parallel with legacy V1 and
- `DelayParams`: Facilitates optional latency simulation for testing.
- **Non-blocking Delay**: Uses `await asyncio.sleep()` to simulate network latency without blocking the Gunicorn worker's event loop.
- **Data-Driven Configuration**: Uses the modern format in `app/ae_obj_types_def.py` to map objects to tables and models.
- **Advanced Search (POST)**: Supports complex, nested filtering via `POST /v3/crud/{obj_type}/search`.
- Recursive AND/OR logic.
- Full operator support: `eq`, `ne`, `gt`, `gte`, `lt`, `lte`, `like`, `in`, `is_null`, `is_not_null`.
- Safe parameterization using unique generated names (e.g., `:sp_1`) to prevent collisions.
## 2. Backward Compatibility Strategy