# engine/_util_cy.py
# Copyright (C) 2010-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
# mypy: disable-error-code="misc, type-arg, untyped-decorator"
from __future__ import annotations

from collections.abc import Mapping
import operator
from typing import Any
from typing import Optional
from typing import Tuple
from typing import TYPE_CHECKING

from .. import exc
from ..util import warn_deprecated

if TYPE_CHECKING:
    from .interfaces import _CoreAnyExecuteParams
    from .interfaces import _CoreMultiExecuteParams
    from .interfaces import _DBAPIAnyExecuteParams
    from .interfaces import _DBAPIMultiExecuteParams
    from .result import _TupleGetterType

# 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

_Empty_Tuple: Tuple[Any, ...] = cython.declare(tuple, ())


@cython.inline
@cython.cfunc
def _is_mapping_or_tuple(value: object, /) -> cython.bint:
    return (
        isinstance(value, dict)
        or isinstance(value, tuple)
        or isinstance(value, Mapping)
        # only do immutabledict or abc.__instancecheck__ for Mapping after
        # we've checked for plain dictionaries and would otherwise raise
    )


@cython.inline
@cython.cfunc
def _is_mapping(value: object, /) -> cython.bint:
    return (
        isinstance(value, dict)
        or isinstance(value, Mapping)
        # only do immutabledict or abc.__instancecheck__ for Mapping after
        # we've checked for plain dictionaries and would otherwise raise
    )


def _distill_params_20(
    params: Optional[_CoreAnyExecuteParams],
) -> _CoreMultiExecuteParams:
    if params is None:
        return _Empty_Tuple
    # Assume list is more likely than tuple
    elif isinstance(params, list) or isinstance(params, tuple):
        # collections_abc.MutableSequence # avoid abc.__instancecheck__
        if len(params) == 0:
            warn_deprecated(
                "Empty parameter sequence passed to execute(). "
                "This use is deprecated and will raise an exception in a "
                "future SQLAlchemy release",
                "2.1",
            )
        elif not _is_mapping(params[0]):
            raise exc.ArgumentError(
                "List argument must consist only of dictionaries"
            )
        return params
    elif _is_mapping(params):
        return [params]  # type: ignore[list-item]
    else:
        raise exc.ArgumentError("mapping or list expected for parameters")


# _is_mapping_or_tuple could be inlined if pure python perf is a problem
def _distill_raw_params(
    params: Optional[_DBAPIAnyExecuteParams],
) -> _DBAPIMultiExecuteParams:
    if params is None:
        return _Empty_Tuple
    elif isinstance(params, list):
        # collections_abc.MutableSequence # avoid abc.__instancecheck__
        if len(params) > 0 and not _is_mapping_or_tuple(params[0]):
            raise exc.ArgumentError(
                "List argument must consist only of tuples or dictionaries"
            )
        return params
    elif _is_mapping_or_tuple(params):
        return [params]  # type: ignore[return-value]
    else:
        raise exc.ArgumentError("mapping or sequence expected for parameters")


@cython.cfunc
def _is_contiguous(indexes: Tuple[int, ...]) -> cython.bint:
    i: cython.Py_ssize_t
    prev: cython.Py_ssize_t
    curr: cython.Py_ssize_t
    for i in range(1, len(indexes)):
        prev = indexes[i - 1]
        curr = indexes[i]
        if prev != curr - 1:
            return False
    return True


def tuplegetter(*indexes: int) -> _TupleGetterType:
    max_index: int
    if len(indexes) == 1 or _is_contiguous(indexes):
        # slice form is faster but returns a list if input is list
        max_index = indexes[-1]
        return operator.itemgetter(slice(indexes[0], max_index + 1))
    else:
        return operator.itemgetter(*indexes)