# sql/_cache_key_cy.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
"""Cython implementation of the cache key traversal.

The traversal for a single class is represented as a :class:`.ClassAttrs`
holding one :class:`.AttrInfo` per attribute that participates in the cache
key.  :class:`.AttrInfo` records the attribute name together with an integer
``kind`` identifying its handler, so that when compiled the per-attribute
dispatch in :meth:`.BaseHasCacheKey._gen_cache_key` is a comparison against a
C integer, rather than the chain of Python ``is`` comparisons against
singleton handler objects that the pure Python traversal loop has to run per
attribute, per cache key.

Traversal kinds that have no inline handler are represented by
:class:`.AttrInfoMethod`, which additionally carries the Python level visit
function to call; these are dispatched under the ``_PY_METHOD`` kind.

Recursion into child elements calls ``_gen_cache_key()`` on the child, which
is this compiled method for the classes that inherit it and the Python level
override otherwise.

"""

# mypy: disable-error-code="untyped-decorator"

from __future__ import annotations

from collections.abc import Sequence
from enum import Enum
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
from typing import TYPE_CHECKING

from .visitors import InternalTraversal
from .. import util

if TYPE_CHECKING:
    from .elements import BindParameter
    from .visitors import _TraverseInternalsType


# START GENERATED CYTHON IMPORT
# This section is automatically generated by the script tools/cython_imports.py
try:
    # NOTE: the cython compiler needs this "import cython" in the file, it
    # can't be only "from sqlalchemy.util import cython" with the fallback
    # in that module
    import cython
except ModuleNotFoundError:
    from sqlalchemy.util import cython


def _is_compiled() -> bool:
    """Utility function to indicate if this module is compiled or not."""
    return cython.compiled  # type: ignore[no-any-return,unused-ignore]


# END GENERATED CYTHON IMPORT

if cython.compiled:
    from cython.cimports.sqlalchemy.sql._util_cy import anon_map
else:
    from ._util_cy import anon_map


class BaseHasCacheKey:
    __slots__ = ()

    @cython.boundscheck(False)
    @cython.wraparound(False)
    def _gen_cache_key(
        self, am: anon_map, bindparams: List[BindParameter[Any]]
    ) -> Optional[Tuple[Any, ...]]:
        """return an optional cache key.

        The cache key is a tuple which can contain any series of
        objects that are hashable and also identifies
        this object uniquely within the presence of a larger SQL expression
        or statement, for the purposes of caching the resulting query.

        The cache key should be based on the SQL compiled structure that would
        ultimately be produced.   That is, two structures that are composed in
        exactly the same way should produce the same cache key; any difference
        in the structures that would affect the SQL string or the type handlers
        should result in a different cache key.

        If a structure cannot produce a useful cache key, the NO_CACHE
        symbol should be added to the anon_map and the method should
        return None.

        """
        cls = self.__class__

        id_: object

        if cython.compiled:
            # inline of anon_map.get_anon(), which otherwise costs a Python
            # level call plus a throwaway (id, found) tuple per node
            am_dict = cython.cast(dict, am)
            idself: object = id(self)
            if idself in am_dict:
                return (am_dict[idself], cls)
            id_ = am._add_missing(idself)
        else:
            id_, found = am.get_anon(self)
            if found:
                return (id_, cls)

        try:
            dispatcher = cls.__dict__["_generated_cache_key_traversal"]
        except KeyError:
            # traversals.py -> _preconfigure_traversals()
            # may be used to run these ahead of time, but
            # is not enabled right now.
            # this block will generate any remaining dispatchers.
            dispatcher = cls._generate_cache_attrs()  # type: ignore[attr-defined]  # noqa: E501

        if dispatcher is _NO_CACHE:
            am[_NO_CACHE] = True
            return None

        disp_cls: ClassAttrs = dispatcher
        result: list[Any] = [id_, cls]
        elements: tuple[Any, ...] = disp_cls.elements
        element: AttrInfo
        py_method: AttrInfoMethod

        for element in elements:
            obj = getattr(self, element.name)
            kind: cython.int = element.kind
            if obj is not None:
                if kind == _STATIC_CACHE_KEY:
                    sck = obj._static_cache_key
                    if sck is _NO_CACHE:
                        am[_NO_CACHE] = True
                        return None
                    result.append(element.name)
                    result.append(sck)
                elif kind == _ANON_NAME:
                    if isinstance(obj, disp_cls.anon_label):
                        obj = obj.apply_map(am)  # type: ignore[attr-defined]
                    result.append(element.name)
                    result.append(obj)
                elif kind == _CALL_GEN_CACHE_KEY:
                    result.append(element.name)
                    result.append(obj._gen_cache_key(am, bindparams))

                # remaining cache functions are against
                # Python tuples, dicts, lists, etc. so we can skip
                # if they are empty
                elif obj:
                    if kind == _CACHE_IN_PLACE:
                        result.append(element.name)
                        result.append(obj)
                    elif kind == _PROPAGATE_ATTRS:
                        result.append(element.name)
                        result.append(obj["compile_state_plugin"])
                        if ps := obj["plugin_subject"]:
                            result.append(ps._gen_cache_key(am, bindparams))
                        else:
                            result.append(None)

                    elif kind == _ANNOTATIONS_KEY:
                        # obj is here is the _annotations dict.  Table uses
                        # a memoized version of it.  however in other cases,
                        # we generate it given am as we may be from a
                        # Join, Aliased, etc.
                        # see #8790

                        if self._gen_static_annotations_cache_key:  # type: ignore[attr-defined]  # noqa: E501
                            result.extend(self._annotations_cache_key)  # type: ignore[attr-defined]  # noqa: E501
                        else:
                            result.extend(self._gen_annotations_cache_key(am))  # type: ignore[attr-defined]  # noqa: E501

                    elif kind == _CLAUSEELEMENT_LIST:
                        result.append(element.name)
                        result.append(
                            tuple(
                                [
                                    elem._gen_cache_key(am, bindparams)
                                    for elem in obj
                                ]
                            ),
                        )
                    elif kind == _PY_METHOD:
                        py_method = element  # type: ignore[assignment]
                        result.extend(
                            py_method.meth(  # type: ignore[operator]
                                py_method.name, obj, self, am, bindparams
                            )
                        )
                    else:
                        raise NotImplementedError(f"Unknown kind {kind}")

        return tuple(result)


class CacheConst(Enum):
    NO_CACHE = 0
    PARAMS = 1


# NOTE: declarations are invisible from python
_NO_CACHE = cython.declare(object, CacheConst.NO_CACHE)
_CACHE_IN_PLACE = cython.declare(cython.const[cython.int], 0)
_CALL_GEN_CACHE_KEY = cython.declare(cython.const[cython.int], 1)
_STATIC_CACHE_KEY = cython.declare(cython.const[cython.int], 2)
_PROPAGATE_ATTRS = cython.declare(cython.const[cython.int], 3)
_ANON_NAME = cython.declare(cython.const[cython.int], 4)
_ANNOTATIONS_KEY = cython.declare(cython.const[cython.int], 5)
_CLAUSEELEMENT_LIST = cython.declare(cython.const[cython.int], 6)
_PY_METHOD = cython.declare(cython.const[cython.int], 7)


class CacheTraverseTarget(Enum):
    """Handlers that the traversal implements inline.

    The value of each member is the integer ``kind`` recorded on the
    :class:`.AttrInfo` for an attribute using that handler, so that the
    traversal can switch on a C integer.

    """

    CACHE_IN_PLACE = _CACHE_IN_PLACE
    CALL_GEN_CACHE_KEY = _CALL_GEN_CACHE_KEY
    STATIC_CACHE_KEY = _STATIC_CACHE_KEY
    PROPAGATE_ATTRS = _PROPAGATE_ATTRS
    ANON_NAME = _ANON_NAME


# the traversal kind for each handler that _generate_class_attrs() may find.
# CacheTraverseTarget members carry their own kind as the enum value; the
# InternalTraversal symbols below have an inline handler but no dedicated
# CacheTraverseTarget member.
_attr_info_kind: Dict[Any, int] = {
    **{member: member.value for member in CacheTraverseTarget},
    InternalTraversal.dp_annotations_key: _ANNOTATIONS_KEY,
    InternalTraversal.dp_clauseelement_list: _CLAUSEELEMENT_LIST,
    InternalTraversal.dp_clauseelement_tuple: _CLAUSEELEMENT_LIST,
    InternalTraversal.dp_memoized_select_entities: _CLAUSEELEMENT_LIST,
}


@cython.cclass
class AttrInfo:
    """An attribute of a specific kind"""

    kind: cython.int
    name: str

    def __init__(self, kind: cython.int, name: str) -> None:
        self.kind = kind
        self.name = name


@cython.cclass
class AttrInfoMethod(AttrInfo):
    """An attribute that requires a python method."""

    meth: object

    def __init__(self, name: str, arg: object) -> None:
        self.kind = _PY_METHOD
        self.name = name
        self.meth = arg


@cython.cclass
class ClassAttrs:
    """The whole cache key traversal for a single class."""

    elements: tuple[AttrInfo, ...]
    anon_label: type

    def __init__(self, elements: Sequence[AttrInfo], anon_label: type) -> None:
        self.elements = tuple(elements)
        self.anon_label = anon_label


class _BaseCacheKeyTraversal:
    """Builds the per-class traversal structure that
    :meth:`.BaseHasCacheKey._gen_cache_key` consumes.

    Mixed in alongside :class:`.HasTraversalDispatch`, which supplies
    :meth:`.dispatch`; the visit symbols themselves are declared by
    :class:`._CacheKeyTraversal` in ``sql/cache_key.py``.

    """

    __slots__ = ()

    if TYPE_CHECKING:

        def dispatch(
            self, visit_symbol: InternalTraversal
        ) -> Callable[..., Any]: ...

    def generate_class_attrs(
        self,
        # NOTE: not annotated as "type": cython enforces annotations on a
        # compiled function and rejects subclasses of builtin types, which
        # fails for any class using a metaclass, e.g. _MetaOptions
        target_cls: Any,
        internal_dispatch: _TraverseInternalsType,
        attrname: str,
    ) -> ClassAttrs:
        """Build the :class:`.ClassAttrs` for ``target_cls`` and memoize it
        on the class under ``attrname``.

        """
        class_attrs = self._generate_class_attrs(internal_dispatch)
        setattr(target_cls, attrname, class_attrs)
        return class_attrs

    @util.preload_module("sqlalchemy.sql.elements")
    def _generate_class_attrs(
        self, internal_dispatch: _TraverseInternalsType
    ) -> ClassAttrs:
        """Resolve the handler for every attribute that participates in the
        cache key for a single class.

        In contrast to :meth:`.HasTraversalDispatch._generate_dispatcher`,
        which emits a function returning a list of
        ``(attrname, value, handler)`` triples that a runtime loop then has to
        interpret, this returns a :class:`.ClassAttrs` holding one
        :class:`.AttrInfo` per attribute, with the handler recorded as an
        integer kind.   Which handler applies to which attribute is a fact
        that's constant for the class, so it's resolved once here rather than
        being rediscovered by a chain of ``is`` comparisons on every cache key
        generation.

        Attributes whose handler has no inline form in the traversal are
        recorded as :class:`.AttrInfoMethod`, which carries the visit method
        itself.

        """
        elems: List[AttrInfo] = []
        meth: Any

        for attrname, visit_sym in internal_dispatch:
            meth = self.dispatch(visit_sym)
            if meth is None:
                continue

            if meth in _attr_info_kind:
                elems.append(AttrInfo(_attr_info_kind[meth], attrname))
            elif isinstance(meth, (CacheTraverseTarget, InternalTraversal)):
                raise NotImplementedError(
                    "No cache key traversal generation rule for symbol "
                    f"{meth!r}, used by attribute {attrname!r}"
                )
            else:
                elems.append(AttrInfoMethod(attrname, meth))

        return ClassAttrs(elems, util.preloaded.sql_elements._anonymous_label)