0ecc5a97d5
In the nested POST handler (api_crud_v3_nested.py), sanitize_payload was running before model instantiation. For secondary FK fields like event_badge_template_id, sanitize_payload resolved the random string → integer, then the model's root_validator stripped the integer back to None (Vision ID anti-leakage guard). Only the parent FK survived because it was explicitly re-injected after serialization. Fix: moved sanitize_payload to run on data_to_insert after serialization, matching the flat V3 POST pattern (api_crud_v3.py). Also moved account_id injection to after sanitize_payload, fixing a latent bug where account_id was silently written as NULL on non-bypass auth. Adds regression test to test_e2e_v3_demo_parity.py that creates an event_badge via nested POST with event_badge_template_id and verifies the field is non-None in the response. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
529 lines
28 KiB
Python
529 lines
28 KiB
Python
from fastapi import APIRouter, Depends, Path, Query, Request, Response, Header
|
|
from pydantic import ValidationError
|
|
from typing import Optional, Union
|
|
import asyncio
|
|
import logging
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
from app.lib_general_v3 import (
|
|
AccountContext, get_account_context,
|
|
PaginationParams, StatusFilterParams,
|
|
SerializationParams, DelayParams
|
|
)
|
|
from app.lib_api_crud_v3 import (
|
|
check_account_access, apply_forced_account_filter, filter_order_by,
|
|
get_supported_filters, safe_json_loads, sanitize_payload, format_db_error,
|
|
apply_vision_id_fix
|
|
)
|
|
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
|
|
|
|
"""
|
|
Aether API V3 - Nested (Child) CRUD Router
|
|
------------------------------------------
|
|
This router handles relational operations between parent and child objects.
|
|
Example: /person/{person_id}/journal/
|
|
"""
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get('/{parent_obj_type}/{parent_obj_id}/{child_obj_type}/', response_model=Resp_Body_Base)
|
|
async def get_child_obj_li(
|
|
response: Response,
|
|
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),
|
|
pagination: PaginationParams = Depends(),
|
|
status_filter: StatusFilterParams = Depends(),
|
|
serialization: SerializationParams = Depends(),
|
|
delay: DelayParams = Depends(),
|
|
):
|
|
"""
|
|
List Child Objects (One-to-Many).
|
|
|
|
Retrieves a list of child objects associated with a specific parent.
|
|
1. Verifies parent existence and user access to the parent.
|
|
2. Filters children where `{parent_obj_type}_id` matches the parent's ID.
|
|
"""
|
|
from app.db_sql import redis_lookup_id_random, sql_select
|
|
import urllib.parse
|
|
|
|
if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
|
|
|
|
qry_dict_li = None
|
|
fulltext_qry_dict_obj = None
|
|
and_qry_dict_obj = None
|
|
and_like_dict_obj = None
|
|
or_like_dict_obj = None
|
|
and_in_dict_li_obj = None
|
|
|
|
jp_obj = safe_json_loads(urllib.parse.unquote(jp)) if jp else None
|
|
if jp_obj:
|
|
if jp_obj.get('qry'): qry_dict_li = jp_obj['qry']
|
|
if jp_obj.get('ft_qry'): fulltext_qry_dict_obj = jp_obj['ft_qry']
|
|
if jp_obj.get('and_qry'): and_qry_dict_obj = jp_obj['and_qry']
|
|
if jp_obj.get('and_like'): and_like_dict_obj = jp_obj['and_like']
|
|
if jp_obj.get('or_like'): or_like_dict_obj = jp_obj['or_like']
|
|
if jp_obj.get('and_in_li'): and_in_dict_li_obj = jp_obj['and_in_li']
|
|
|
|
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=f"Invalid object type(s).")
|
|
|
|
# ID Vision: Resolve physical table names from registry to support aliases
|
|
parent_table = obj_type_kv_li[parent_obj_type].get('tbl')
|
|
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')))
|
|
|
|
# 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.")
|
|
|
|
order_by_li = filter_order_by(order_by_li, base_name, table_name)
|
|
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)
|
|
|
|
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,
|
|
qry_dict_li=qry_dict_li,
|
|
fulltext_qry_dict=fulltext_qry_dict_obj,
|
|
and_qry_dict=and_qry_dict_obj,
|
|
and_like_dict=and_like_dict_obj,
|
|
or_like_dict=or_like_dict_obj,
|
|
and_in_dict_li=and_in_dict_li_obj,
|
|
order_by_li=order_by_li,
|
|
limit=pagination.limit,
|
|
offset=pagination.offset,
|
|
as_list=True,
|
|
)
|
|
|
|
if sql_result is False:
|
|
# Standardized rich error bubbling
|
|
db_err = format_db_error(get_last_sql_error())
|
|
|
|
# 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 = [apply_vision_id_fix(base_name(**record).dict(by_alias=serialization.by_alias, exclude_unset=serialization.exclude_unset, exclude_defaults=serialization.exclude_defaults, exclude_none=serialization.exclude_none), obj_name, serialization.by_alias) 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).")
|
|
|
|
# ID Vision: Resolve physical table names from registry to support aliases
|
|
parent_table = obj_type_kv_li[parent_obj_type].get('tbl')
|
|
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_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
|
|
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:
|
|
resp_data_li = [apply_vision_id_fix(base_name(**record).dict(by_alias=serialization.by_alias, exclude_unset=serialization.exclude_unset, exclude_defaults=serialization.exclude_defaults, exclude_none=serialization.exclude_none), obj_name, serialization.by_alias) 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}/', response_model=Resp_Body_Base)
|
|
async def post_child_obj(
|
|
request: Request,
|
|
response: Response,
|
|
parent_obj_type: str = Path(min_length=2, max_length=50),
|
|
parent_obj_id: str = Path(min_length=11, max_length=22),
|
|
child_obj_type: str = Path(min_length=2, max_length=50),
|
|
return_obj: Optional[bool] = True,
|
|
x_ae_ignore_extra_fields: Optional[bool] = Header(False),
|
|
account: AccountContext = Depends(get_account_context),
|
|
serialization: SerializationParams = Depends(),
|
|
delay: DelayParams = Depends(),
|
|
):
|
|
"""
|
|
Create Child Object.
|
|
|
|
1. Verifies Parent existence and access.
|
|
2. Automatically links the new child to the parent (`{parent_obj_type}_id` = parent_id).
|
|
3. Performs standard creation logic (validation, injection, sanitization).
|
|
"""
|
|
from app.db_sql import redis_lookup_id_random, sql_select, sql_insert, get_id_random
|
|
|
|
if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
|
|
|
|
obj_data = await request.json()
|
|
|
|
if parent_obj_type not in obj_type_kv_li or child_obj_type not in obj_type_kv_li:
|
|
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid object type.")
|
|
|
|
# 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'))
|
|
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.")
|
|
|
|
try:
|
|
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:
|
|
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)
|
|
|
|
# Sanitize AFTER serialization so that:
|
|
# 1. The model receives raw Vision ID strings (passes field-length constraints).
|
|
# 2. ID resolution (string → integer) happens on the dict going to the DB,
|
|
# avoiding the root_validator's integer-stripping anti-leakage guard.
|
|
# (Matches the flat V3 POST pattern in api_crud_v3.py.)
|
|
sanitize_payload(data_to_insert, input_model, ignore_extra=x_ae_ignore_extra_fields)
|
|
|
|
# Enforce account ownership AFTER sanitize_payload so the integer account_id goes
|
|
# straight to the DB without conflicting with Vision ID string constraints in the model.
|
|
if not account.super and account.auth_method != 'bypass' and account.account_id:
|
|
if 'account_id' in input_model.__fields__:
|
|
data_to_insert['account_id'] = account.account_id
|
|
|
|
# Re-inject parent FK last — overrides anything sanitize_payload or the model may have
|
|
# set — ensuring the child is always linked to the correct parent.
|
|
data_to_insert[f'{parent_obj_type}_id'] = resolved_parent_id
|
|
|
|
if sql_insert_result := sql_insert(data=data_to_insert, table_name=table_name_insert):
|
|
new_obj_id = sql_insert_result
|
|
new_obj_id_random = get_id_random(record_id=new_obj_id, table_name=child_obj_type)
|
|
|
|
if return_obj:
|
|
if sql_select_result := sql_select(table_name=table_name_select, record_id=new_obj_id):
|
|
resp_data = output_model(**sql_select_result).dict(by_alias=serialization.by_alias, exclude_unset=serialization.exclude_unset)
|
|
apply_vision_id_fix(resp_data, child_obj_type, serialization.by_alias)
|
|
return mk_resp(data=resp_data, response=response)
|
|
return mk_resp(data={"obj_id": new_obj_id_random, "obj_id_random": new_obj_id_random}, response=response)
|
|
else:
|
|
# Standardized rich error bubbling
|
|
db_err = format_db_error(get_last_sql_error())
|
|
return mk_resp(data=False, status_code=400, response=response, status_message="Failed to create child object.", details=db_err.dict())
|
|
|
|
|
|
@router.get('/{parent_obj_type}/{parent_obj_id}/{child_obj_type}/{child_obj_id}', response_model=Resp_Body_Base)
|
|
async def get_child_obj(
|
|
response: Response,
|
|
parent_obj_type: str = Path(min_length=2, max_length=50),
|
|
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(),
|
|
):
|
|
"""
|
|
Retrieve Child Object.
|
|
|
|
Verifies that the child belongs to the specified parent.
|
|
"""
|
|
from app.db_sql import redis_lookup_id_random, sql_select
|
|
|
|
if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
|
|
|
|
# ID Vision: Resolve physical table names from registry to support aliases
|
|
if parent_obj_type not in obj_type_kv_li or child_obj_type not in obj_type_kv_li:
|
|
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid object type(s).")
|
|
|
|
parent_table = obj_type_kv_li[parent_obj_type].get('tbl')
|
|
child_table = obj_type_kv_li[child_obj_type].get('tbl')
|
|
|
|
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_table)
|
|
resolved_child_id = redis_lookup_id_random(record_id_random=child_obj_id, table_name=child_table)
|
|
|
|
if not resolved_parent_id or not resolved_child_id:
|
|
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')))
|
|
|
|
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:
|
|
return mk_resp(data=False, status_code=404, response=response, status_message="Child not found under parent.")
|
|
|
|
resp_data = base_name(**sql_result).dict(by_alias=serialization.by_alias, exclude_unset=serialization.exclude_unset)
|
|
apply_vision_id_fix(resp_data, child_obj_type, serialization.by_alias)
|
|
return mk_resp(data=resp_data, response=response)
|
|
return mk_resp(data=False, status_code=404, response=response, status_message="Child not found.")
|
|
|
|
|
|
@router.patch('/{parent_obj_type}/{parent_obj_id}/{child_obj_type}/{child_obj_id}', response_model=Resp_Body_Base)
|
|
async def patch_child_obj(
|
|
request: Request,
|
|
response: Response,
|
|
parent_obj_type: str = Path(min_length=2, max_length=50),
|
|
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),
|
|
return_obj: Optional[bool] = True,
|
|
x_ae_ignore_extra_fields: Optional[bool] = Header(False),
|
|
account: AccountContext = Depends(get_account_context),
|
|
serialization: SerializationParams = Depends(),
|
|
delay: DelayParams = Depends(),
|
|
):
|
|
"""
|
|
Update Child Object.
|
|
|
|
Verifies that the child belongs to the specified parent before updating.
|
|
"""
|
|
from app.db_sql import redis_lookup_id_random, sql_select, sql_update
|
|
|
|
if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
|
|
|
|
obj_data = await request.json()
|
|
|
|
# ID Vision: Resolve physical table names from registry to support aliases
|
|
if parent_obj_type not in obj_type_kv_li or child_obj_type not in obj_type_kv_li:
|
|
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid object type(s).")
|
|
|
|
parent_table = obj_type_kv_li[parent_obj_type].get('tbl')
|
|
child_table = obj_type_kv_li[child_obj_type].get('tbl')
|
|
|
|
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_table)
|
|
resolved_child_id = redis_lookup_id_random(record_id_random=child_obj_id, table_name=child_table)
|
|
|
|
if not resolved_parent_id or not resolved_child_id:
|
|
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_update = 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'))
|
|
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.get(f'{parent_obj_type}_id') != resolved_parent_id:
|
|
return mk_resp(data=False, status_code=404, response=response, status_message="Child not found under parent.")
|
|
else:
|
|
return mk_resp(data=False, status_code=404, response=response, status_message="Child not found.")
|
|
|
|
# Sanitize payload (ID resolution, virtual fields, and optionally extra fields)
|
|
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 return_obj:
|
|
if updated_child := sql_select(table_name=table_name_select, record_id=resolved_child_id):
|
|
resp_data = output_model(**updated_child).dict(by_alias=serialization.by_alias, exclude_unset=serialization.exclude_unset)
|
|
apply_vision_id_fix(resp_data, child_obj_type, serialization.by_alias)
|
|
return mk_resp(data=resp_data, response=response)
|
|
return mk_resp(data=True, response=response, status_message="Updated successfully.")
|
|
else:
|
|
db_err = format_db_error(get_last_sql_error())
|
|
return mk_resp(data=False, status_code=400, response=response, status_message="Update failed.", details=db_err.dict())
|
|
|
|
|
|
|
|
@router.delete('/{parent_obj_type}/{parent_obj_id}/{child_obj_type}/{child_obj_id}', response_model=Resp_Body_Base)
|
|
async def delete_child_obj(
|
|
response: Response,
|
|
parent_obj_type: str = Path(min_length=2, max_length=50),
|
|
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),
|
|
method: str = Query('delete', regex='^(delete|hide|disable)$'),
|
|
account: AccountContext = Depends(get_account_context),
|
|
delay: DelayParams = Depends(),
|
|
):
|
|
"""
|
|
Delete Child Object.
|
|
|
|
Verifies that the child belongs to the specified parent before deleting.
|
|
"""
|
|
from app.db_sql import redis_lookup_id_random, sql_select, sql_update, sql_delete
|
|
|
|
if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
|
|
|
|
# ID Vision: Resolve physical table names from registry to support aliases
|
|
if parent_obj_type not in obj_type_kv_li or child_obj_type not in obj_type_kv_li:
|
|
return mk_resp(data=False, status_code=400, response=response, status_message="Invalid object type(s).")
|
|
|
|
parent_table = obj_type_kv_li[parent_obj_type].get('tbl')
|
|
child_table = obj_type_kv_li[child_obj_type].get('tbl')
|
|
|
|
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_table)
|
|
resolved_child_id = redis_lookup_id_random(record_id_random=child_obj_id, table_name=child_table)
|
|
|
|
if not resolved_parent_id or not resolved_child_id:
|
|
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_delete = obj_cfg.get('tbl_update', obj_cfg.get('tbl'))
|
|
table_name_select = obj_cfg.get('tbl_default', obj_cfg.get('tbl'))
|
|
|
|
if existing_child := sql_select(table_name=table_name_select, record_id=resolved_child_id):
|
|
if existing_child.get(f'{parent_obj_type}_id') != resolved_parent_id:
|
|
return mk_resp(data=False, status_code=404, response=response, status_message="Child not found under parent.")
|
|
else:
|
|
return mk_resp(data=False, status_code=404, response=response, status_message="Child not found.")
|
|
|
|
if method == 'hide':
|
|
success = sql_update(table_name=table_name_delete, record_id=resolved_child_id, data={'hide': True})
|
|
elif method == 'disable':
|
|
success = sql_update(table_name=table_name_delete, record_id=resolved_child_id, data={'enable': False})
|
|
else:
|
|
success = sql_delete(table_name=table_name_delete, record_id=resolved_child_id)
|
|
|
|
if success:
|
|
return mk_resp(data=True, response=response, status_message=f"Deleted successfully via {method}.")
|
|
return mk_resp(data=False, status_code=400, response=response, status_message="Deletion failed.") |