feat(data-store): add advanced search and delay support to V3 code lookup

This commit is contained in:
Scott Idem
2026-02-03 14:51:04 -05:00
parent fde729d474
commit 03154ab95b

View File

@@ -146,6 +146,7 @@ async def get_v3_data_store_obj_w_code(
account: AccountContext = Depends(get_account_context),
serialization: SerializationParams = Depends(),
status_filter: StatusFilterParams = Depends(),
delay: DelayParams = Depends(),
):
"""
V3 Standardized Data Store Lookup.
@@ -162,12 +163,13 @@ async def get_v3_data_store_obj_w_code(
response=Response()
)
return handle_get_data_store_obj_w_code(
return await handle_get_data_store_obj_w_code(
data_store_code = data_store_code,
for_type = for_type,
for_id = for_id,
commons = v3_commons,
limit = limit,
delay = delay,
)
# ### END ### API Data Store ### get_v3_data_store_obj_w_code() ###
@@ -224,6 +226,52 @@ async def get_data_store_obj_w_code_query(
)
from app.models.api_crud_models import SearchFilter, SearchQuery
@router.post('/v3/data_store/code/{data_store_code}/search', response_model=Resp_Body_Base, tags=['Data Store V3'])
async def search_v3_data_store_obj_w_code(
data_store_code: str,
search_query: SearchQuery,
for_type: Optional[str] = Query(None),
for_id: Optional[str] = Query(None),
view: str = Query('default'),
account: AccountContext = Depends(get_account_context),
pagination: PaginationParams = Depends(),
status_filter: StatusFilterParams = Depends(),
delay: DelayParams = Depends(),
):
"""
Advanced Search for Data Store (within a code hierarchy).
"""
if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
# 1. Resolve IDs
resolved_for_id = None
if for_type and for_id:
resolved_for_id = redis_lookup_id_random(record_id_random=for_id, table_name=for_type)
if not resolved_for_id:
return mk_resp(data=False, status_code=404, status_message="Parent for_id not found.")
# 2. Apply advanced search on top of the standard lookup
# We use sql_select with the SearchQuery
sql_result = sql_select(
table_name='data_store',
enabled=status_filter.enabled,
hidden=status_filter.hidden,
search_query=search_query,
field_name='code',
field_value=data_store_code,
limit=pagination.limit,
offset=pagination.offset,
as_list=True,
)
if sql_result is False:
return mk_resp(data=False, status_code=500, status_message="Database error during search.")
return mk_resp(data=sql_result)
def handle_get_data_store_obj_w_code(
data_store_code: str,
for_type: Optional[str],