Source code for edri.integrations.sqlalchemy.path

from __future__ import annotations

from typing import Any

from edri.utility.function import camel2snake
from sqlalchemy import inspect as sqlalchemy_inspect

__all__ = ["SQLAlchemyPathError", "canonical_orm_path"]




class SQLAlchemyPathError(ValueError):
    """Raised when a public dotted path does not match mapped ORM attributes."""





def canonical_orm_path(
    model: type[Any],
    public_path: str,
    *,
    require_scalar: bool = False,
) -> tuple[str, ...]:
    """Resolve a camelCase dotted path to strict mapped attribute names.

    Every non-final segment must be a relationship. A column may only occur as
    the final segment. When ``require_scalar`` is true, a relationship may not
    be the final segment either.
    """

    if not isinstance(public_path, str) or not public_path.strip():
        raise SQLAlchemyPathError("ORM path cannot be empty")

    public_segments = public_path.split(".")
    current_model = model
    result: list[str] = []

    for index, public_segment in enumerate(public_segments):
        segment = public_segment.strip()
        if not segment:
            raise SQLAlchemyPathError(f'Invalid ORM path "{public_path}"')

        mapper = sqlalchemy_inspect(current_model)
        attribute = camel2snake(segment)
        if attribute == "id" and len(mapper.primary_key) == 1:
            attribute = mapper.primary_key[0].key
        is_last = index == len(public_segments) - 1

        if attribute in mapper.column_attrs:
            if not is_last:
                raise SQLAlchemyPathError(
                    f'Cannot traverse scalar attribute "{attribute}" in "{public_path}"'
                )
            result.append(attribute)
            continue

        relationship = mapper.relationships.get(attribute)
        if relationship is None:
            raise SQLAlchemyPathError(
                f'Unknown mapped attribute "{attribute}" in "{public_path}"'
            )
        if is_last and require_scalar:
            raise SQLAlchemyPathError(
                f'ORM path "{public_path}" ends in a relationship'
            )

        result.append(attribute)
        current_model = relationship.mapper.class_

    return tuple(result)