Rust Bridge Internals¶

def model_fields(model: Type[ModelType]) -> dict[str, FieldMetadata]:
    """Return Ormdantic field metadata for a Pydantic model."""
    return {
        name: FieldMetadata(name=name, info=field)
        for name, field in model.model_fields.items()
    }

def model_field(model: Type[ModelType], name: str) -> FieldMetadata:
    """Return field metadata for one field on a Pydantic model."""
    return FieldMetadata(name=name, info=model.model_fields[name])

def annotation_allows_none(annotation: Any) -> bool:
    """Return whether an annotation accepts `None`."""
    if annotation is NoneType:
        return True
    return NoneType in get_args(annotation)

def is_union_annotation(annotation: Any) -> bool:
    """Return whether an annotation is a PEP 604 or typing union."""
    return get_origin(annotation) in {UnionType, Union}

def is_list_annotation(annotation: Any) -> bool:
    """Return whether an annotation is a concrete list annotation."""
    return get_origin(annotation) is list

def contains_list_annotation(annotation: Any) -> bool:
    """Return whether an annotation contains a list at any nesting level."""
    if is_list_annotation(annotation):
        return True
    return any(contains_list_annotation(arg) for arg in get_args(annotation))

def is_dict_annotation(annotation: Any) -> bool:
    """Return whether an annotation is a dictionary."""
    return get_origin(annotation) is dict or annotation is dict

def first_model_arg(
    annotation: Any, table_models: set[type[BaseModel]]
) -> type[BaseModel] | None:
    """Return the first registered table model found in an annotation."""
    try:
        if annotation in table_models:
            return annotation
    except TypeError:
        pass
    for arg in get_args(annotation):
        try:
            if arg in table_models:
                return arg
        except TypeError:
            pass
        if nested := first_model_arg(arg, table_models):
            return nested
    return None

def rebuild_model(model: Type[ModelType]) -> None:
    """Rebuild a Pydantic model after forward references are available."""
    model.model_rebuild()

def validate_table_map(table_map: Map) -> int | None:
    """Validate current Python table metadata through Rust when available."""
    if _ormdantic is None or not hasattr(_ormdantic, "validate_schema_tables"):
        return None
    tables = [
        (table.tablename, table.pk, list(table.columns))
        for table in table_map.name_to_data.values()
    ]
    return int(_ormdantic.validate_schema_tables(tables))

def compile_create_table_sql(table_map: Map, tablename: str, dialect: str) -> list[str]:
    """Compile create-table DDL statements for a registered table."""
    rust = _require_schema_symbol("compile_create_table_sql")
    table = table_map.name_to_data[tablename]
    columns = [
        column_descriptor(table_map, table, field_name, field)
        for field_name, field in model_fields(table.model).items()
        if field_name not in table.back_references
    ]
    return list(
        rust.compile_create_table_sql(
            dialect,
            table.tablename,
            columns,
            index_descriptors(table),
            table.unique_constraints,
        )
    )

def compile_drop_table_sql(tablename: str, dialect: str) -> str:
    """Compile a drop-table DDL statement for a table name."""
    rust = _require_schema_symbol("compile_drop_table_sql")
    return str(rust.compile_drop_table_sql(dialect, tablename))

def hydrate_flat_payload(
    *,
    tablename: str,
    pk: str,
    columns: list[str],
    rows: list[tuple[Any, ...]],
    is_array: bool,
) -> dict[str, Any] | list[dict[str, Any]] | None:
    """Hydrate flat SQL rows into dict payloads."""
    if _ormdantic is not None:
        return cast(
            dict[str, Any] | list[dict[str, Any]] | None,
            _ormdantic.hydrate_flat(
                tablename,
                pk,
                columns,
                [list(row) for row in rows],
                is_array,
            ),
        )
    return _hydrate_flat_payload_python(
        tablename=tablename,
        pk=pk,
        columns=columns,
        rows=rows,
        is_array=is_array,
    )

def hydrate_joined_payload(
    *,
    columns: list[str],
    rows: list[tuple[Any, ...]],
    path_pks: list[tuple[str, str]],
    array_paths: list[str],
) -> dict[str, Any] | None:
    """Hydrate joined SQL rows into nested dict payloads."""
    if _ormdantic is None or not hasattr(_ormdantic, "hydrate_joined"):
        raise RuntimeError(
            "Ormdantic requires the Rust extension for joined hydration. "
            "Install the package with maturin or reinstall the wheel."
        )
    payload = cast(
        dict[str, Any],
        _ormdantic.hydrate_joined(
            columns,
            [list(row) for row in rows],
            path_pks,
            array_paths,
        ),
    )
    return payload or None

def plan_result_shape(
    *,
    root_table: str,
    columns: list[str],
    array_paths: list[str] | None = None,
) -> dict[str, Any]:
    """Describe joined result aliases with Rust when available."""
    array_paths = array_paths or []
    if _ormdantic is not None and hasattr(_ormdantic, "plan_result_shape"):
        return cast(
            dict[str, Any],
            _ormdantic.plan_result_shape(root_table, columns, array_paths),
        )
    return _plan_result_shape_python(root_table, columns, array_paths)