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

Add INFO logging for nested parent-resolution and add E2E nested-create test for event_badge

Browse Source
This commit is contained in:
Scott Idem
2026-03-27 11:30:08 -04:00
parent 687472f4e3
commit 7f87f32b70
2 changed files with 131 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
app/routers/api_crud_v3_nested.py
View File

@@ -85,6 +85,9 @@ async def get_child_obj_li(
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')))
# Log parent/child resolution details (use INFO so logs appear in production)
log.info("nested.list start parent=%s parent_table=%s parent_id_random=%s child=%s table=%s allowed_parents=%s", parent_obj_type, parent_table, parent_obj_id, obj_name, table_name, obj_cfg.get('parent_types'))
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration error.")
@@ -92,15 +95,26 @@ async def get_child_obj_li(
status_filter = get_supported_filters(base_name, status_filter)
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_table)
log.info("nested.list resolved_parent_id=%s (random=%s) for parent_table=%s", resolved_parent_id, parent_obj_id, parent_table)
if not resolved_parent_id:
log.info("nested.list parent resolution failed for random id=%s table=%s", parent_obj_id, parent_table)
return mk_resp(data=False, status_code=404, response=response, status_message=f"Parent not found.")
# Enforce allowed parent types when configured on the child object
allowed_parents = obj_cfg.get('parent_types')
if allowed_parents and parent_obj_type not in allowed_parents:
log.info("nested.list invalid parent type: parent=%s allowed=%s", parent_obj_type, allowed_parents)
return mk_resp(data=False, status_code=400, response=response, status_message=f"Invalid parent type for this child.")
parent_cfg = obj_type_kv_li[parent_obj_type]
parent_table_select = parent_cfg.get('tbl_default', parent_cfg.get('tbl'))
if parent_sql_res := sql_select(table_name=parent_table_select, record_id=resolved_parent_id):
log.info("nested.list parent_sql_res found for id=%s table=%s", resolved_parent_id, parent_table_select)
if not check_account_access(parent_sql_res, account, parent_obj_type):
log.info("nested.list access denied to parent id=%s for account=%s", resolved_parent_id, getattr(account, 'account_id', None))
return mk_resp(data=False, status_code=403, response=response, status_message="Access denied to parent.")
else:
log.info("nested.list parent sql_select returned no row for id=%s table=%s", resolved_parent_id, parent_table_select)
return mk_resp(data=False, status_code=404, response=response, status_message="Parent not found.")
and_qry_dict_obj = apply_forced_account_filter(and_qry_dict_obj, account, base_name, obj_name, table_name=table_name)
@@ -182,15 +196,26 @@ async def search_child_obj_li(
searchable_fields = obj_cfg.get('searchable_fields')
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_table)
log.info("nested.search resolved_parent_id=%s (random=%s) for parent_table=%s", resolved_parent_id, parent_obj_id, parent_table)
if not resolved_parent_id:
log.info("nested.search parent resolution failed for random id=%s table=%s", parent_obj_id, parent_table)
return mk_resp(data=False, status_code=404, response=response, status_message="Parent not found.")
# Enforce allowed parent types when configured on the child object
allowed_parents = obj_cfg.get('parent_types')
if allowed_parents and parent_obj_type not in allowed_parents:
log.info("nested.search invalid parent type: parent=%s allowed=%s", parent_obj_type, allowed_parents)
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid parent type for this child.")
parent_cfg = obj_type_kv_li[parent_obj_type]
parent_table_select = parent_cfg.get('tbl_default', parent_cfg.get('tbl'))
if parent_sql_res := sql_select(table_name=parent_table_select, record_id=resolved_parent_id):
log.info("nested.search parent_sql_res found for id=%s table=%s", resolved_parent_id, parent_table_select)
if not check_account_access(parent_sql_res, account, parent_obj_type):
log.info("nested.search access denied to parent id=%s for account=%s", resolved_parent_id, getattr(account, 'account_id', None))
return mk_resp(data=False, status_code=403, response=response, status_message="Access denied to parent.")
else:
log.info("nested.search parent sql_select returned no row for id=%s table=%s", resolved_parent_id, parent_table_select)
return mk_resp(data=False, status_code=404, response=response, status_message="Parent not found.")
# Enforce account isolation on the search query
@@ -257,18 +282,29 @@ async def post_child_obj(
# ID Vision: Resolve physical table names from registry to support aliases
parent_table = obj_type_kv_li[parent_obj_type].get('tbl')
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_table)
log.info("nested.post parent=%s parent_table=%s parent_id_random=%s", parent_obj_type, parent_table, parent_obj_id)
log.info("nested.post resolved_parent_id=%s for random=%s table=%s", resolved_parent_id, parent_obj_id, parent_table)
if not resolved_parent_id:
log.info("nested.post parent resolution failed for random id=%s table=%s", parent_obj_id, parent_table)
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_select = parent_cfg.get('tbl_default', parent_cfg.get('tbl'))
if parent_sql_res := sql_select(table_name=parent_table_select, record_id=resolved_parent_id):
log.info("nested.post parent_sql_res found for id=%s table=%s", resolved_parent_id, parent_table_select)
if not check_account_access(parent_sql_res, account, parent_obj_type):
log.info("nested.post access denied to parent id=%s for account=%s", resolved_parent_id, getattr(account, 'account_id', None))
return mk_resp(data=False, status_code=403, response=response, status_message="Access denied to parent.")
else:
log.info("nested.post parent sql_select returned no row for id=%s table=%s", resolved_parent_id, parent_table_select)
return mk_resp(data=False, status_code=404, response=response, status_message="Parent not found.")
obj_cfg = obj_type_kv_li[child_obj_type]
# Enforce allowed parent types when configured on the child object
allowed_parents = obj_cfg.get('parent_types')
if allowed_parents and parent_obj_type not in allowed_parents:
log.info("nested.post invalid parent type: parent=%s allowed=%s", parent_obj_type, allowed_parents)
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid parent type for this child.")
table_name_insert = obj_cfg.get('tbl_update', obj_cfg.get('tbl'))
table_name_select = obj_cfg.get('tbl_default', obj_cfg.get('tbl'))
input_model = obj_cfg.get('mdl_in', obj_cfg.get('mdl'))
@@ -351,6 +387,10 @@ 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]
# Enforce allowed parent types when configured on the child object
allowed_parents = obj_cfg.get('parent_types')
if allowed_parents and parent_obj_type not in allowed_parents:
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid parent type for this child.")
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')))