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

feat(api-v3): add advanced search and view support to nested CRUD router

Browse Source
This commit is contained in:
Scott Idem
2026-02-03 14:10:41 -05:00
parent 9e423806df
commit 9362938ffe
2 changed files with 171 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
app/routers/api_crud_v3_nested.py
View File

@@ -17,6 +17,7 @@ from app.lib_api_crud_v3 import (
)
from app.db_sql import get_last_sql_error
from app.models.response_models import *
from app.models.api_crud_models import SearchFilter, SearchQuery
from app.ae_obj_types_def import obj_type_kv_li
"""
@@ -34,6 +35,7 @@ async def get_child_obj_li(
parent_obj_type: str,
parent_obj_id: str,
child_obj_type: str,
view: str = Query('default'),
order_by_li: Optional[str] = None,
jp: Optional[Union[str, None]] = None,
account: AccountContext = Depends(get_account_context),
@@ -77,8 +79,8 @@ async def get_child_obj_li(
return mk_resp(data=False, status_code=400, response=response, status_message=f"Invalid object type(s).")
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'))
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
base_name = obj_cfg.get(f'mdl_{view}', 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 error.")
@@ -125,6 +127,90 @@ async def get_child_obj_li(
# If it's a schema error (like Unknown Column), it's a 400 Bad Request
status_code = 400 if db_err.category == "database_schema" else 500
return mk_resp(data=False, status_code=status_code, response=response, status_message="Listing failed due to database error.", details=db_err.dict())
if sql_result:
resp_data_li = [base_name(**record).dict(by_alias=serialization.by_alias, exclude_unset=serialization.exclude_unset, exclude_defaults=serialization.exclude_defaults, exclude_none=serialization.exclude_none) for record in sql_result]
return mk_resp(data=resp_data_li, response=response)
else:
return mk_resp(data=[], status_code=200, response=response)
@router.post('/{parent_obj_type}/{parent_obj_id}/{child_obj_type}/search', response_model=Resp_Body_Base, tags=['CRUD v3 Search (Dev)'])
async def search_child_obj_li(
response: Response,
parent_obj_type: str,
parent_obj_id: str,
child_obj_type: str,
search_query: SearchQuery,
view: str = Query('default'),
order_by_li: Optional[str] = Query(None),
account: AccountContext = Depends(get_account_context),
pagination: PaginationParams = Depends(),
status_filter: StatusFilterParams = Depends(),
serialization: SerializationParams = Depends(),
delay: DelayParams = Depends(),
):
"""
Search Child Objects (POST).
Advanced search endpoint for nested objects.
"""
from app.db_sql import redis_lookup_id_random, sql_select
if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
order_by_li = safe_json_loads(order_by_li)
obj_name = child_obj_type
if obj_name not in obj_type_kv_li or parent_obj_type not in obj_type_kv_li:
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid object type(s).")
obj_cfg = obj_type_kv_li[obj_name]
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
base_name = obj_cfg.get(f'mdl_{view}', 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 error.")
order_by_li = filter_order_by(order_by_li, base_name, table_name)
status_filter = get_supported_filters(base_name, status_filter)
searchable_fields = obj_cfg.get('searchable_fields')
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_obj_type)
if not resolved_parent_id:
return mk_resp(data=False, status_code=404, response=response, status_message="Parent not found.")
parent_cfg = obj_type_kv_li[parent_obj_type]
parent_table = parent_cfg.get('tbl_default', parent_cfg.get('tbl'))
if parent_sql_res := sql_select(table_name=parent_table, record_id=resolved_parent_id):
if not check_account_access(parent_sql_res, account, parent_obj_type):
return mk_resp(data=False, status_code=403, response=response, status_message="Access denied to parent.")
else:
return mk_resp(data=False, status_code=404, response=response, status_message="Parent not found.")
# Enforce account isolation on the search query
if not account.super and account.auth_method != 'bypass' and account.account_id:
if search_query.and_filters is None: search_query.and_filters = []
if 'account_id' in base_name.__fields__:
search_query.and_filters.append(SearchFilter(field='account_id', op='eq', value=account.account_id))
sql_result = sql_select(
table_name=table_name,
field_name=f'{parent_obj_type}_id',
field_value=resolved_parent_id,
enabled=status_filter.enabled,
hidden=status_filter.hidden,
search_query=search_query,
searchable_fields=searchable_fields,
order_by_li=order_by_li,
limit=pagination.limit,
offset=pagination.offset,
as_list=True,
)
if sql_result is False:
db_err = format_db_error(get_last_sql_error())
status_code = 400 if db_err.category == "database_schema" else 500
return mk_resp(data=False, status_code=status_code, response=response, status_message="Search failed due to database error.", details=db_err.dict())
if sql_result:
@@ -225,6 +311,7 @@ async def get_child_obj(
parent_obj_id: str = Path(min_length=11, max_length=22),
child_obj_type: str = Path(min_length=2, max_length=50),
child_obj_id: str = Path(min_length=11, max_length=22),
view: str = Query('default'),
account: AccountContext = Depends(get_account_context),
serialization: SerializationParams = Depends(),
delay: DelayParams = Depends(),
@@ -245,8 +332,8 @@ async def get_child_obj(
return mk_resp(data=False, status_code=404, response=response, status_message="Object(s) not found.")
obj_cfg = obj_type_kv_li[child_obj_type]
table_name = obj_cfg.get('tbl_default', obj_cfg.get('tbl'))
base_name = obj_cfg.get('mdl_default', obj_cfg.get('mdl'))
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
base_name = obj_cfg.get(f'mdl_{view}', obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
if sql_result := sql_select(table_name=table_name, record_id=resolved_child_id):
if sql_result.get(f'{parent_obj_type}_id') != resolved_parent_id:

80
tests/e2e/test_e2e_v3_nested_advanced.py Normal file
View File

@@ -0,0 +1,80 @@
import requests
import json
# Configuration
BASE_URL = "https://dev-api.oneskyit.com/v3/crud"
API_KEY = "PMM4n50teUCaOMMTN8qOJA"
ACCOUNT_ID = "Q8lR8Ai8hx2FjbQ3C_EH1Q" # OSIT
def test_nested_search():
print("--- Testing Nested Advanced Search (POST) ---")
# Test: Search for journals belonging to a specific person
parent_type = "person"
parent_id = "--ghJX-ztEM" # Using a valid person ID
child_type = "journal"
url = f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/search"
headers = {
"X-Aether-API-Key": API_KEY,
"x-account-id": ACCOUNT_ID
}
search_query = {
"and_filters": [
{"field": "name", "op": "like", "value": "%"}
]
}
try:
response = requests.post(url, headers=headers, json=search_query)
print(f"Status: {response.status_code}")
if response.status_code == 200:
data = response.json().get("data", [])
print(f"✅ Success: Found {len(data)} nested records via search.")
return True
else:
print(f"❌ Failed: {response.text}")
return False
except Exception as e:
print(f"💥 Exception: {e}")
return False
def test_nested_view():
print("\n--- Testing Nested Get with View Parameter ---")
# Test: Get a single journal entry under a journal using 'enriched' view
parent_type = "journal"
parent_id = "PJRCGHQWERT" # Using a known journal ID
child_type = "journal_entry"
child_id = "PJRCGHQWERT" # Using a known entry ID
url = f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/{child_id}?view=enriched"
headers = {
"X-Aether-API-Key": API_KEY,
"x-account-id": ACCOUNT_ID
}
try:
response = requests.get(url, headers=headers)
print(f"Status: {response.status_code}")
if response.status_code == 200:
print("✅ Success: Retrieved nested record with view=enriched.")
return True
elif response.status_code == 404:
print("⚠️ Note: Record not found, but route matched.")
return True
else:
print(f"❌ Failed: {response.status_code}")
return False
except Exception as e:
print(f"💥 Exception: {e}")
return False
if __name__ == "__main__":
s1 = test_nested_search()
s2 = test_nested_view()
if s1 and s2:
print("\n🎉 NESTED V3 FEATURES VERIFIED!")