Adapters

class BaseAdapter(ABC):
    """Abstract base class for data adapters.

    Defines the contract for all data operations (get, list, create, update, delete).
    Subclass this to add support for a new ORM or data source.

    Example:
        ```python
        class MyAdapter(BaseAdapter):
            async def get(self, pk): ...
        ```
    """

    model: Any
    #: Primary-key codec for ``model``. ORM adapters replace it with an introspected
    #: value; third-party adapters inherit the historical ``int id`` default.
    pk: PrimaryKeyInfo = DEFAULT_PK

    def __init__(
        self, model: Any, engine: Any = None, *, session_factory: SessionFactory | None = None
    ) -> None:
        require_engine_or_session_factory(type(self).__name__, engine, session_factory)
        self.model = model
        self.engine = engine
        self.session_factory = session_factory
        self._queryset_filter: QuerysetFilter | None = None

    def _session(self) -> AbstractAsyncContextManager[AsyncSessionLike]:
        """Open a session from whichever of ``engine`` / ``session_factory`` was given.

        Prefers ``session_factory`` (the host's own session maker) when set, so a
        host-managed engine, connection pool and transaction scope are respected.
        Falls back to opening a plain ``AsyncSession`` on ``engine`` for the legacy,
        engine-only construction path.
        """
        return open_session(self.engine, self.session_factory)

    def for_model(self, model: Any) -> BaseAdapter:
        """Return a new adapter of this same class for ``model``.

        Carries over this adapter's ``engine`` and ``session_factory`` so the new
        adapter shares the same database connection seam — used to construct inline
        / related-model adapters instead of reaching for a concrete adapter class
        and ``self.engine`` directly (which breaks a ``session_factory=``-only
        admin).

        ``session_factory`` is forwarded only when set, so a third-party
        ``BaseAdapter`` subclass that still uses the pre-0.5.8, engine-only
        ``__init__(self, model, engine)`` signature keeps working unchanged
        (it never sees an unexpected ``session_factory`` keyword).
        """
        kwargs: dict[str, Any] = {"engine": self.engine}
        if self.session_factory is not None:
            kwargs["session_factory"] = self.session_factory
        return type(self)(model, **kwargs)

    def datetime_kind(self, field: str) -> DateTimeKind | None:
        """Return whether ``field`` stores timezone-aware or naive datetimes.

        The default reads the pydantic annotation only: ``AwareDatetime`` is
        ``"aware"``, any other datetime is ``"naive"`` and non-datetime or unknown
        fields return ``None``. ORM adapters override this with column introspection.
        """
        try:
            model_fields = schema_for(self.model).model_fields
        except TypeError:  # a model class the schema registry does not know
            return None
        if field not in model_fields:
            return None
        return annotation_datetime_kind(model_fields[field].annotation)

    def get_queryset(self, request: Request | None = None) -> dict[str, Any]:
        """Return additional equality filters merged into ``list()`` and ``get()`` queries.

        The returned mapping has the shape ``{column_name: value}`` and is applied as
        equality predicates **before** any view-layer filters. Use this hook to implement
        features like row-level security or tenant scoping without leaking those concerns
        into the view layer.

        Subclasses may override this method directly, or callers may register a
        per-request callable via :meth:`set_queryset_filter`. The callable is
        re-evaluated on every ``list()`` / ``get()`` — no caching — so it must be pure
        or request-scoped.

        Default behaviour returns an empty dict, which is a no-op (backward compatible).

        Args:
            request: The active Starlette/FastAPI request, when available. ``None`` is
                permitted so the hook is callable from contexts without an HTTP request
                (e.g. management scripts, tests).

        Returns:
            A dict of column-name to value pairs to be ANDed into the query's WHERE
            clause. Returns ``{}`` by default.
        """
        if self._queryset_filter is None:
            return {}
        return self._queryset_filter(request)

    def set_queryset_filter(self, filter_fn: QuerysetFilter) -> None:
        """Register a per-request callable that returns additional queryset filters.

        The callable receives the current :class:`starlette.requests.Request` (or
        ``None``) and must return a dict of ``{column_name: value}`` equality filters.
        It is re-invoked on every ``list()`` / ``get()`` so it may consult
        request-scoped state (e.g. ``request.state.user``).

        Args:
            filter_fn: A callable taking an optional :class:`Request` and returning a
                dict of equality filters.
        """
        self._queryset_filter = filter_fn

    def _resolve_queryset_filters(self, request: Request | None = None) -> dict[str, Any]:
        """Invoke :meth:`get_queryset` and validate the return type.

        Raises:
            TypeError: If ``get_queryset`` returns a value that is not a ``dict``.
        """
        result = self.get_queryset(request)
        if not isinstance(result, dict):
            raise TypeError(
                f"{type(self).__name__}.get_queryset() must return a dict, "
                f"got {type(result).__name__}"
            )
        scoped = scoped_queryset_filters(self.model, request)
        if not scoped:
            return result
        return {**result, **scoped}

    async def ensure_inline_rows_owned(
        self,
        spec: InlineModelSpec,
        rows: builtins.list[dict[str, Any]],
        parent_pk: Any,
    ) -> None:
        """Raise :class:`InlineRowNotOwnedError` unless every ``_pk`` is a child of the parent.

        Must run before any write. A submitted child pk is never trusted on its
        own: it must be an existing child (``fk_field == parent_pk``) of the
        parent being saved. Adapters provide the lookup via :meth:`inline_child_pks`,
        which returns DB-typed keys (e.g. ``uuid.UUID`` for a UUID-keyed child).

        A submitted ``_pk`` is a key already parsed by ``InlineFormset`` with the
        child's codec, or whatever a direct adapter caller passes. It is parsed
        (again) through the *child* model's own primary-key codec (not the
        parent's) before comparing against ``owned``, so a
        UUID-keyed child's submitted ``str(uuid)`` compares equal to the
        ``uuid.UUID`` instances ``inline_child_pks`` returns, instead of every
        row with an existing ``_pk`` being rejected as foreign.
        """
        submitted_raw = [row["_pk"] for row in rows if row.get("_pk") is not None]
        if not submitted_raw:
            return
        child_pk = self.for_model(spec.model).pk

        def _parse_or_foreign(raw: Any) -> tuple[Any, Any]:
            # Cannot possibly be a real child key — reject it as foreign
            # rather than letting the parse error bubble up as a 500.
            try:
                return (child_pk.parse(raw), _FOREIGN)
            except InvalidPrimaryKey:
                return (_FOREIGN, raw)

        parsed_or_foreign = [_parse_or_foreign(raw) for raw in submitted_raw]
        submitted = {parsed for parsed, _ in parsed_or_foreign if parsed is not _FOREIGN}
        foreign = {raw for _, raw in parsed_or_foreign if raw is not _FOREIGN}
        owned = await self.inline_child_pks(spec, parent_pk) if parent_pk is not None else set()
        foreign |= submitted - owned
        if foreign:
            raise InlineRowNotOwnedError(
                f"Inline rows {sorted(map(str, foreign))} do not belong to this parent"
            )

    async def inline_child_pks(self, spec: InlineModelSpec, parent_pk: Any) -> set[Any]:  # noqa: ARG002
        """Return the primary keys of ``spec.model`` rows whose FK points at ``parent_pk``.

        Adapters that support inline formsets override this. The default vouches
        for no key, so any submitted child pk is rejected.
        """
        return set()

    @abstractmethod
    async def get(self, pk: Any) -> Any:
        """
        Retrieves a single object by its primary key.

        Args:
            pk: The primary key of the object to retrieve.

        Returns:
            The retrieved object, or None if not found.
        """
        raise NotImplementedError

    @abstractmethod
    async def list(
        self,
        page: int = 1,
        page_size: int = 10,
        search: str | None = None,
        filters: Mapping[str, Any] | Sequence[FilterCondition] | None = None,
        order_by: str | None = None,
        search_fields: builtins.list[str] | None = None,
    ) -> tuple[builtins.list[Any], int]:
        """
        Retrieves a list of objects with optional pagination, searching, and filtering.

        Args:
            page: The page number for pagination.
            page_size: The number of items per page.
            search: A search query to filter the results.
            filters: Either the legacy ``{column: value}`` equality mapping, or a
                ``Sequence[FilterCondition]`` (``core/filtering.py``) carrying
                ``exact``/``gte``/``lte``/``in``/``isnull`` predicates.
                Implementations resolve either form through
                :func:`hyperadmin.core.filtering.normalize_filters`.
            order_by: The field to order the results by.

        Returns:
            A tuple containing the list of objects and the total count of objects.
        """
        raise NotImplementedError

    @abstractmethod
    async def create(self, data: dict[str, Any]) -> Any:
        """
        Creates a new object.

        Args:
            data: A dictionary of data for the new object.

        Returns:
            The created object.
        """
        raise NotImplementedError

    @abstractmethod
    async def update(self, pk: Any, data: dict[str, Any]) -> Any:
        """
        Updates an existing object.

        Args:
            pk: The primary key of the object to update.
            data: A dictionary of data to update the object with.

        Returns:
            The updated object.
        """
        raise NotImplementedError

    @abstractmethod
    async def delete(self, pk: Any) -> None:
        """
        Deletes an object by its primary key.

        Args:
            pk: The primary key of the object to delete.
        """
        raise NotImplementedError

    @abstractmethod
    async def get_related(self, pk: Any, field: str) -> builtins.list[Any]:
        """
        Retrieves related objects for a given object and field.

        Args:
            pk: The primary key of the object.
            field: The name of the relationship field.

        Returns:
            A list of related objects.
        """
        raise NotImplementedError

    @abstractmethod
    async def get_schema(self) -> dict[str, Any]:
        """
        Retrieves the schema definition for the model.

        Returns:
            A dictionary representing the model's schema.
        """
        raise NotImplementedError

    @abstractmethod
    async def get_choices(
        self,
        field: str,
        q: str = "",
        limit: int = 50,
        offset: int = 0,
        label_fn: Callable[[Any], str] | None = None,
        **filters: Any,
    ) -> builtins.list[ChoiceItem]:
        """Return paginated, searchable choices for a relation field.

        Args:
            field: The relation field name on this adapter's model.
            q: Optional search string (ILIKE match on string columns).
            limit: Max results to return. Must not exceed 200.
            offset: Number of rows to skip for pagination.
            label_fn: Renders each target row's option label (``relation_display``);
                defaults to :func:`~hyperadmin.core.display.get_display_name`.
            **filters: Extra equality filters forwarded to the query (cascading support).

        Returns:
            A list of ``ChoiceItem`` dicts with ``value``, ``label``, ``selected=False``.

        Raises:
            ValueError: When ``limit`` exceeds 200.
        """
        raise NotImplementedError

    @abstractmethod
    async def save_inline_rows(
        self,
        spec: InlineModelSpec,
        rows: builtins.list[dict[str, Any]],
        parent_pk: Any,
    ) -> None:
        """Persist validated inline rows — create, update, or delete as needed.

        Args:
            spec: The ``InlineModelSpec`` describing the related model and FK field.
            rows: Validated row dicts, each optionally containing ``_pk`` (for
                update/delete) and ``_delete`` (for deletion).
            parent_pk: The primary key of the parent object to associate new rows with.
        """
        raise NotImplementedError

class SQLModelAdapter(BaseAdapter):
    """
    Data adapter for SQLModel.
    """

    def __init__(
        self,
        model: type[SQLModel],
        engine: AsyncEngine | None = None,
        *,
        session_factory: SessionFactory | None = None,
    ):
        super().__init__(model, engine, session_factory=session_factory)
        self.inspector = inspect(model)
        self.pk = introspect_primary_key(model)

    def datetime_kind(self, field: str) -> "DateTimeKind | None":
        """Return the column-introspected datetime kind for ``field``.

        Overrides :meth:`BaseAdapter.datetime_kind` (annotation-only) with the
        full rule set in :func:`hyperadmin.adapters.introspection.datetime_kind`,
        which also inspects the mapped column type.
        """
        try:
            return _introspect_datetime_kind(self.model, field)
        except KeyError:
            return None

    async def get(self, pk: Any) -> Any:
        """
        Retrieves a single object by its primary key.

        Any filters returned by :meth:`get_queryset` are merged into the WHERE clause
        before the primary-key predicate so excluded rows resolve to ``None``.

        Args:
            pk: The primary key of the object to retrieve.

        Returns:
            The retrieved object, or ``None`` if not found, or if ``pk`` does not
            parse as this model's key type.

        Raises:
            NotImplementedError: For a composite primary key.
        """
        try:
            parsed_pk = self.pk.parse(pk)
        except InvalidPrimaryKey:
            return None
        queryset_filters = self._resolve_queryset_filters()
        async with self._session() as session:
            mapper = inspect(self.model)
            options = [selectinload(getattr(self.model, rel.key)) for rel in mapper.relationships]
            query = (
                select(self.model)
                .where(getattr(self.model, self.pk.attr) == parsed_pk)
                .options(*options)
            )
            for key, value in queryset_filters.items():
                query = query.where(getattr(self.model, key) == value)
            result = await session.execute(query)
            return result.unique().scalar_one_or_none()

    async def list(
        self,
        page: int = 1,
        page_size: int = 10,
        search: str | None = None,
        filters: Mapping[str, Any] | Sequence[FilterCondition] | None = None,
        order_by: str | None = None,
        search_fields: list[str] | None = None,
    ) -> tuple[list[Any], int]:
        """
        Retrieves a list of objects with optional pagination, searching, and filtering.

        Filters returned by :meth:`get_queryset` are applied **before** any
        view-layer ``filters`` so they cannot be bypassed. The same predicates are
        included in the count query, keeping pagination math consistent.

        ``filters`` accepts either the legacy ``{column: value}`` equality
        mapping or a ``Sequence[FilterCondition]``; both are resolved through
        :func:`hyperadmin.core.filtering.normalize_filters` and compiled to
        clauses via :mod:`hyperadmin.adapters._filter_clause`, shared with
        :class:`~hyperadmin.adapters.sqlalchemy.SQLAlchemyAdapter`.
        """
        queryset_filters = self._resolve_queryset_filters()
        async with self._session() as session:
            query = select(self.model)

            # Eagerly load all relationships to avoid DetachedInstanceError
            mapper = inspect(self.model)
            for rel in mapper.relationships:
                query = query.options(selectinload(getattr(self.model, rel.key)))

            # Apply queryset hook filters first (RLS / tenant scoping)
            for key, value in queryset_filters.items():
                query = query.where(getattr(self.model, key) == value)

            # Apply filtering: a legacy dict or a Sequence[FilterCondition],
            # normalized to typed conditions and compiled to clauses shared
            # with SQLAlchemyAdapter.
            for clause in build_clauses(self.model, normalize_filters(filters)):
                query = query.where(clause)

            # Apply searching using configured search_fields
            if search:
                fields_to_search = (
                    search_fields if search_fields is not None else self._detect_search_fields()
                )
                if fields_to_search:
                    conditions = []
                    for field_name in fields_to_search:
                        col = getattr(self.model, field_name, None)
                        if col is not None:
                            conditions.append(col.ilike(f"%{search}%"))
                    if conditions:
                        query = query.where(or_(*conditions))

            # Apply ordering
            if order_by:
                if order_by.startswith("-"):
                    query = query.order_by(getattr(self.model, order_by[1:]).desc())
                else:
                    query = query.order_by(getattr(self.model, order_by).asc())

            # Get total count
            count_query = select(func.count()).select_from(query.subquery())
            total_count_result = await session.execute(count_query)
            total_count = total_count_result.scalar_one()

            # Apply pagination
            offset = (page - 1) * page_size
            query = query.offset(offset).limit(page_size)

            # Get the rows
            results = await session.execute(query)
            return list(results.unique().scalars().all()), total_count

    def _detect_search_fields(self) -> builtins.list[str]:
        """Detect non-sensitive string columns on the model for search fallback."""
        return detect_search_columns(self.model, self.inspector)

    async def create(self, data: dict[str, Any]) -> Any:
        """
        Creates a new object.

        Args:
            data: A dictionary of data for the new object.

        Returns:
            The created object.
        """
        async with self._session() as session:
            db_obj = self.model(**data)
            session.add(db_obj)
            await session.commit()
            await session.refresh(db_obj)
            return db_obj

    async def update(self, pk: Any, data: dict[str, Any]) -> Any:
        """
        Updates an existing object.

        Primary-key attributes in ``data`` are dropped before assignment, so the
        key is always immutable through this method.

        Args:
            pk: The primary key of the object to update.
            data: A dictionary of data to update the object with.

        Returns:
            The updated object.

        Raises:
            InvalidPrimaryKey: If ``pk`` does not parse as this model's key type.
            NotImplementedError: For a composite primary key.
        """
        parsed_pk = self.pk.parse(pk)
        data = {key: value for key, value in data.items() if key not in self.pk.attrs}
        async with self._session() as session:
            db_obj = await session.get(self.model, parsed_pk)
            if db_obj:
                for key, value in data.items():
                    setattr(db_obj, key, value)
                session.add(db_obj)
                await session.commit()
                await session.refresh(db_obj)
            return db_obj

    async def delete(self, pk: Any) -> None:
        """
        Deletes an object.

        Args:
            pk: The primary key of the object to delete.

        Raises:
            InvalidPrimaryKey: If ``pk`` does not parse as this model's key type.
            NotImplementedError: For a composite primary key.
        """
        parsed_pk = self.pk.parse(pk)
        async with self._session() as session:
            db_obj = await session.get(self.model, parsed_pk)
            if db_obj:
                await session.delete(db_obj)
                await session.commit()

    async def get_related(self, pk: Any, field: str) -> builtins.list[Any]:
        """
        Retrieves related objects for a given object and field.

        Args:
            pk: The primary key of the object.
            field: The name of the related field.

        Returns:
            A list of related objects, or ``[]`` if not found, or if ``pk`` does
            not parse as this model's key type.

        Raises:
            NotImplementedError: For a composite primary key.
        """
        try:
            parsed_pk = self.pk.parse(pk)
        except InvalidPrimaryKey:
            return []
        queryset_filters = self._resolve_queryset_filters()
        async with self._session() as session:
            query = (
                select(self.model)
                .where(getattr(self.model, self.pk.attr) == parsed_pk)
                .options(selectinload(getattr(self.model, field)))
            )
            for key, value in queryset_filters.items():
                query = query.where(getattr(self.model, key) == value)
            result = await session.execute(query)
            db_obj = result.unique().scalar_one_or_none()
            if db_obj:
                return getattr(db_obj, field)
            return []

    async def get_schema(self) -> dict[str, Any]:
        """
        Returns the JSON schema for the model.

        Returns:
            A dictionary representing the JSON schema.
        """
        return self.model.model_json_schema()

    async def get_choices(
        self,
        field: str,
        q: str = "",
        limit: int = 50,
        offset: int = 0,
        label_fn: Callable[[Any], str] | None = None,
        **filters: Any,
    ) -> builtins.list[ChoiceItem]:
        """Return paginated, searchable choices for a relation field.

        Performs a single SELECT on the related model — no N+1.
        """
        if limit > _MAX_CHOICES_LIMIT:
            raise ValueError(f"limit {limit} exceeds maximum of {_MAX_CHOICES_LIMIT}")

        mapper = inspect(self.model)
        target_model = None
        for rel in mapper.relationships:
            if rel.key == field:
                target_model = rel.mapper.class_
                break

        if target_model is None:
            return []

        target_inspector = inspect(target_model)
        target_pk = introspect_primary_key(target_model)
        async with self._session() as session:
            query = select(target_model)

            if q:
                # Never match on sensitive columns (q would be a substring oracle).
                str_cols = [
                    getattr(target_model, name)
                    for name in detect_search_columns(target_model, target_inspector)
                ]
                if str_cols:
                    query = query.where(or_(*[c.ilike(f"%{q}%") for c in str_cols[:3]]))

            # Row scoping of the target model's admin (tenant / RLS) always applies.
            for key, value in scoped_queryset_filters(target_model).items():
                query = query.where(getattr(target_model, key) == value)

            # Cascade filters: the view forwards only keys declared by the widget.
            target_sensitive = effective_sensitive_field_names(target_model)
            for key, value in filters.items():
                if hasattr(target_model, key) and key not in target_sensitive:
                    try:
                        coerced = coerce_column_value(target_inspector, key, value)
                    except ValueError:
                        # An unparseable cascade value (e.g. '' or 'garbage' for a
                        # UUID FK) matches no row rather than raising a 500 — '' is
                        # what a dependent select sends while its parent is blank.
                        return []
                    query = query.where(getattr(target_model, key) == coerced)

            query = query.offset(offset).limit(limit)
            result = await session.execute(query)
            items = result.unique().scalars().all()

        return [
            ChoiceItem(
                # The target's own key codec, not a hardcoded "id": canonical for
                # any key kind (UUIDs become lowercase, hyphenated strings).
                value=target_pk.to_str(target_pk.value_of(item)),
                # Never str(item): SQLModel's default __str__ prints every column.
                label=label_fn(item) if label_fn else get_display_name(item),
                selected=False,
            )
            for item in items
        ]

    async def save_inline_rows(
        self,
        spec: InlineModelSpec,
        rows: builtins.list[dict[str, Any]],
        parent_pk: Any,
    ) -> None:
        """Persist validated inline rows — create, update, or delete as needed.

        A fresh ``SQLModelAdapter`` is constructed for the inline model via
        :meth:`for_model`, so it shares this adapter's ``engine`` or
        ``session_factory``.

        Args:
            spec: The ``InlineModelSpec`` describing the related model and FK field.
            rows: Validated row dicts, each optionally containing ``_pk`` (for
                update/delete) and ``_delete`` (for deletion).
            parent_pk: The primary key of the parent object to associate new rows with.

        Raises:
            InlineRowNotOwnedError: Before any write, when a submitted ``_pk`` is not an
                existing child of ``parent_pk``.
        """
        await self.ensure_inline_rows_owned(spec, rows, parent_pk)
        inline_adapter = self.for_model(spec.model)
        for row in rows:
            row_pk = row.get("_pk")
            if row.get("_delete") and row_pk is not None:
                await inline_adapter.delete(pk=row_pk)
            elif row_pk is not None:
                # Keys are immutable and the row stays attached to its parent;
                # inline_adapter.update() drops any pk attrs from row_data itself.
                row_data = {
                    k: v for k, v in row.items() if k not in ("_pk", "_delete", spec.fk_field)
                }
                await inline_adapter.update(pk=row_pk, data=row_data)
            elif not row.get("_delete"):
                row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
                row_data[spec.fk_field] = parent_pk
                await inline_adapter.create(data=row_data)

    async def inline_child_pks(self, spec: InlineModelSpec, parent_pk: Any) -> set[Any]:
        """Return the primary keys of ``spec.model`` rows owned by ``parent_pk``."""
        pk_col = inspect(spec.model).primary_key[0]
        async with self._session() as session:
            result = await session.execute(
                select(pk_col).where(getattr(spec.model, spec.fk_field) == parent_pk)
            )
            return set(result.scalars().all())