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

Permissive Update: Implement x-ae-ignore-extra-fields header support for nested routes

Browse Source
This commit is contained in:
Scott Idem
2026-01-19 16:48:48 -05:00
parent a269e2a716
commit 19e64135ca
1 changed files with 16 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
app/routers/api_crud_v3_nested.py
View File

@@ -1,4 +1,5 @@
from fastapi import APIRouter, Depends, Path, Query, Request, Response from fastapi import APIRouter, Depends, Path, Query, Request, Response, Header
from pydantic import ValidationError
from typing import Optional, Union from typing import Optional, Union
import asyncio import asyncio
import logging import logging
@@ -132,6 +133,7 @@ async def post_child_obj(
parent_obj_id: str = Path(min_length=11, max_length=22), parent_obj_id: str = Path(min_length=11, max_length=22),
child_obj_type: str = Path(min_length=2, max_length=50), child_obj_type: str = Path(min_length=2, max_length=50),
return_obj: Optional[bool] = True, return_obj: Optional[bool] = True,
x_ae_ignore_extra_fields: Optional[bool] = Header(False),
account: AccountContext = Depends(get_account_context), account: AccountContext = Depends(get_account_context),
serialization: SerializationParams = Depends(), serialization: SerializationParams = Depends(),
delay: DelayParams = Depends(), delay: DelayParams = Depends(),
@@ -170,22 +172,28 @@ async def post_child_obj(
input_model = obj_cfg.get('mdl_in', obj_cfg.get('mdl')) input_model = obj_cfg.get('mdl_in', obj_cfg.get('mdl'))
output_model = obj_cfg.get('mdl_out', obj_cfg.get('mdl_default', obj_cfg.get('mdl'))) output_model = obj_cfg.get('mdl_out', obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
if not table_name_insert or not input_model or not table_name_select or not output_model:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration error.")
if not account.super and account.auth_method != 'bypass' and account.account_id: if not account.super and account.auth_method != 'bypass' and account.account_id:
if 'account_id' in input_model.__fields__: if 'account_id' in input_model.__fields__:
obj_data['account_id'] = account.account_id obj_data['account_id'] = account.account_id
obj_data[f'{parent_obj_type}_id'] = resolved_parent_id obj_data[f'{parent_obj_type}_id'] = resolved_parent_id
# Sanitize payload (ID resolution, virtual fields, and optionally extra fields)
sanitize_payload(obj_data, input_model, ignore_extra=x_ae_ignore_extra_fields)
try: try:
validated_obj = input_model(**obj_data) validated_obj = input_model(**obj_data)
except ValidationError as e:
structured_errors = {err['loc'][-1]: err['msg'] for err in e.errors()}
return mk_resp(data=False, status_code=400, response=response, status_message="Validation Failed", details=structured_errors)
except Exception as e: except Exception as e:
return mk_resp(data=False, status_code=400, response=response, status_message="Validation Failed", details=str(e)) return mk_resp(data=False, status_code=400, response=response, status_message="Validation Failed", details=str(e))
data_to_insert = validated_obj.dict(exclude_unset=True) data_to_insert = validated_obj.dict(exclude_unset=True)
# Sanitize payload (remove virtual fields and view-only fields)
sanitize_payload(data_to_insert, input_model)
if sql_insert_result := sql_insert(data=data_to_insert, table_name=table_name_insert): if sql_insert_result := sql_insert(data=data_to_insert, table_name=table_name_insert):
new_obj_id = sql_insert_result new_obj_id = sql_insert_result
new_obj_id_random = get_id_random(record_id=new_obj_id, table_name=child_obj_type) new_obj_id_random = get_id_random(record_id=new_obj_id, table_name=child_obj_type)
@@ -248,6 +256,7 @@ async def patch_child_obj(
child_obj_type: str = Path(min_length=2, max_length=50), child_obj_type: str = Path(min_length=2, max_length=50),
child_obj_id: str = Path(min_length=11, max_length=22), child_obj_id: str = Path(min_length=11, max_length=22),
return_obj: Optional[bool] = True, return_obj: Optional[bool] = True,
x_ae_ignore_extra_fields: Optional[bool] = Header(False),
account: AccountContext = Depends(get_account_context), account: AccountContext = Depends(get_account_context),
serialization: SerializationParams = Depends(), serialization: SerializationParams = Depends(),
delay: DelayParams = Depends(), delay: DelayParams = Depends(),
@@ -271,6 +280,7 @@ async def patch_child_obj(
obj_cfg = obj_type_kv_li[child_obj_type] obj_cfg = obj_type_kv_li[child_obj_type]
table_name_update = obj_cfg.get('tbl_update', obj_cfg.get('tbl')) table_name_update = obj_cfg.get('tbl_update', obj_cfg.get('tbl'))
table_name_select = obj_cfg.get('tbl_default', 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'))
output_model = obj_cfg.get('mdl_out', obj_cfg.get('mdl_default', obj_cfg.get('mdl'))) output_model = obj_cfg.get('mdl_out', obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
if existing_child := sql_select(table_name=table_name_select, record_id=resolved_child_id): if existing_child := sql_select(table_name=table_name_select, record_id=resolved_child_id):
@@ -279,8 +289,8 @@ async def patch_child_obj(
else: else:
return mk_resp(data=False, status_code=404, response=response, status_message="Child not found.") return mk_resp(data=False, status_code=404, response=response, status_message="Child not found.")
# Sanitize payload (remove virtual fields and view-only fields) # Sanitize payload (ID resolution, virtual fields, and optionally extra fields)
sanitize_payload(obj_data, output_model) sanitize_payload(obj_data, input_model, ignore_extra=x_ae_ignore_extra_fields)
if sql_update(data=obj_data, table_name=table_name_update, record_id=resolved_child_id): if sql_update(data=obj_data, table_name=table_name_update, record_id=resolved_child_id):
if return_obj: if return_obj: