OSIT-AE-API-FastAPI/app/lib_api_crud_v3.py

from typing import Any, Dict, Optional
import json
import logging

from app.lib_general_v3 import AccountContext, StatusFilterParams

log = logging.getLogger(__name__)

def check_account_access(sql_result: Any, account: AccountContext, obj_name: str = None) -> bool:
    """
    Enforce Multi-Tenant Data Isolation.
    
    Verifies that the requested record belongs to the authenticated user's account.
    Returns True if:
    - User is a Super User or System (Bypass).
    - The record's `account_id` matches the user's `account_id`.
    """
    if account.super or account.auth_method == 'bypass':
        return True
    if not account.account_id:
        return False
    
    res_account_id = None
    if isinstance(sql_result, dict):
        if obj_name == 'account':
            res_account_id = sql_result.get('id')
        else:
            res_account_id = sql_result.get('account_id')
    
    if res_account_id is not None and res_account_id != account.account_id:
        return False
    return True

def apply_forced_account_filter(and_qry_dict: Optional[Dict], account: AccountContext, model: Any, obj_name: str) -> Dict:
    """
    Secure Search Filtering.
    
    Automatically appends an `account_id` filter to database queries to ensure 
    users only retrieve records associated with their own account.
    """
    forced = and_qry_dict or {}
    if account.super or account.auth_method == 'bypass':
        return forced
    
    if obj_name == 'account':
        forced['id'] = account.account_id
    elif model and hasattr(model, '__fields__') and 'account_id' in model.__fields__:
        forced['account_id'] = account.account_id
        
    return forced

def filter_order_by(order_by_li: Any, model: Any, table_name: str = None) -> Optional[Dict[str, str]]:
    """
    Sanitize Sorting Parameters.
    
    Prevents SQL injection and logic errors by validating that requested sort columns
    actually exist in the Pydantic model and/or the database table.
    """
    if not order_by_li or not isinstance(order_by_li, dict) or not model:
        return order_by_li
    if not hasattr(model, '__fields__'):
        return order_by_li
    
    model_fields = set(model.__fields__.keys())
    model_fields.update({f.alias for f in model.__fields__.values() if f.alias})
    filtered = {k: v for k, v in order_by_li.items() if k in model_fields}
    
    if table_name and filtered:
        from app.db_sql import db
        from sqlalchemy import text
        final_filtered = {}
        for column in filtered:
            try:
                # Lightweight check to see if column exists in SQL
                db.execute(text(f"SELECT `{column}` FROM `{table_name}` LIMIT 0"))
                final_filtered[column] = filtered[column]
            except Exception:
                pass
        filtered = final_filtered
    return filtered

def get_supported_filters(model: Any, status_filter: StatusFilterParams) -> StatusFilterParams:
    """
    Adaptive Status Filtering.
    
    Adjusts the default filters (enabled/hidden) based on whether the target object
    actually supports those concepts (i.e., has those columns).
    """
    if not model or not hasattr(model, "__fields__"):
        return status_filter
    # We create a new instance to avoid side effects on the dependency object
    from app.routers.dependencies_v3 import StatusFilterParams as SF
    adjusted = SF()
    adjusted.enabled = status_filter.enabled
    adjusted.hidden = status_filter.hidden
    
    if 'enable' not in model.__fields__:
        adjusted.enabled = 'all'
    if 'hide' not in model.__fields__:
        adjusted.hidden = 'all'
    return adjusted

def safe_json_loads(json_str: Optional[str]) -> Any:
    if not json_str or json_str == 'undefined': return None
    try: return json.loads(json_str)
    except: return None

def sanitize_payload(data: dict, model: Any) -> None:
    """
    Sanitizes an input payload before database insertion or update.
    
    1. Removes virtual lookup fields (ending in `_id_random`) that are used for API
       convenience but do not exist in the database.
    2. Removes fields explicitly marked for exclusion in the model's 
       `fields_to_exclude_from_db` ClassVar (e.g., view-only fields).
       
    Modifies the `data` dictionary in-place.
    """
    if not isinstance(data, dict):
        return

    # Filter out virtual _id_random fields (e.g., account_id_random)
    keys_to_remove = [k for k in data.keys() if k.endswith('_id_random') and k != 'id_random']
    for k in keys_to_remove:
        del data[k]

    # Filter out model-specific excluded fields (e.g., view-only fields)
    if hasattr(model, 'fields_to_exclude_from_db'):
        for k in model.fields_to_exclude_from_db:
             if k in data:
                 del data[k]