Function flatten_pydantic_instance

langroid/utils/pydantic_utils.py:207–251

Given a possibly nested Pydantic instance, return a flattened version of it, as a dict where nested traversal paths are translated to keys a__b__c. Args: instance (BaseModel): The Pydantic instance to flatten. prefix (str, optional): The prefix to use for the top-level

(
    instance: BaseModel,
    prefix: str = "",
    force_str: bool = False,
)

def flatten_pydantic_instance(
    instance: BaseModel,
    prefix: str = "",
    force_str: bool = False,
) -> Dict[str, Any]:
    """
    Given a possibly nested Pydantic instance, return a flattened version of it,
    as a dict where nested traversal paths are translated to keys a__b__c.

    Args:
        instance (BaseModel): The Pydantic instance to flatten.
        prefix (str, optional): The prefix to use for the top-level fields.
        force_str (bool, optional): Whether to force all values to be strings.

    Returns:
        Dict[str, Any]: The flattened dict.

    """
    flat_data: Dict[str, Any] = {}
    for name, value in instance.model_dump().items():
        # Assuming nested pydantic model will be a dict here
        if isinstance(value, dict):
            # Get field info from model_fields
            field_info = instance.model_fields[name]
            # Try to get the nested model type from field annotation
            field_type = (
                field_info.annotation if hasattr(field_info, "annotation") else None
            )
            if (
                field_type
                and isinstance(field_type, type)
                and issubclass(field_type, BaseModel)
            ):
                nested_flat_data = flatten_pydantic_instance(
                    field_type(**value),
                    prefix=f"{prefix}{name}__",
                    force_str=force_str,
                )
            else:
                # Skip non-Pydantic nested fields for safety
                continue
            flat_data.update(nested_flat_data)
        else:
            flat_data[f"{prefix}{name}"] = str(value) if force_str else value
    return flat_data


def extract_fields(doc: BaseModel, fields: List[str]) -> Dict[str, Any]: