Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/sql»

venv/Lib/site-packages/sqlalchemy/sql/_selectable_constructors.py

@@ -0,0 +1,763 @@
+# sql/_selectable_constructors.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
+
+from __future__ import annotations
+
+from typing import Any
+from typing import Optional
+from typing import overload
+from typing import Tuple
+from typing import TYPE_CHECKING
+from typing import Union
+
+from . import coercions
+from . import roles
+from ._typing import _ColumnsClauseArgument
+from ._typing import _no_kw
+from .elements import ColumnClause
+from .selectable import Alias
+from .selectable import CompoundSelect
+from .selectable import Exists
+from .selectable import FromClause
+from .selectable import Join
+from .selectable import Lateral
+from .selectable import LateralFromClause
+from .selectable import NamedFromClause
+from .selectable import Select
+from .selectable import TableClause
+from .selectable import TableSample
+from .selectable import Values
+
+if TYPE_CHECKING:
+    from ._typing import _FromClauseArgument
+    from ._typing import _OnClauseArgument
+    from ._typing import _OnlyColumnArgument
+    from ._typing import _SelectStatementForCompoundArgument
+    from ._typing import _T0
+    from ._typing import _T1
+    from ._typing import _T2
+    from ._typing import _T3
+    from ._typing import _T4
+    from ._typing import _T5
+    from ._typing import _T6
+    from ._typing import _T7
+    from ._typing import _T8
+    from ._typing import _T9
+    from ._typing import _TP
+    from ._typing import _TypedColumnClauseArgument as _TCCA
+    from .functions import Function
+    from .selectable import CTE
+    from .selectable import HasCTE
+    from .selectable import ScalarSelect
+    from .selectable import SelectBase
+
+
+def alias(
+    selectable: FromClause, name: Optional[str] = None, flat: bool = False
+) -> NamedFromClause:
+    """Return a named alias of the given :class:`.FromClause`.
+
+    For :class:`.Table` and :class:`.Join` objects, the return type is the
+    :class:`_expression.Alias` object. Other kinds of :class:`.NamedFromClause`
+    objects may be returned for other kinds of :class:`.FromClause` objects.
+
+    The named alias represents any :class:`_expression.FromClause` with an
+    alternate name assigned within SQL, typically using the ``AS`` clause when
+    generated, e.g. ``SELECT * FROM table AS aliasname``.
+
+    Equivalent functionality is available via the
+    :meth:`_expression.FromClause.alias`
+    method available on all :class:`_expression.FromClause` objects.
+
+    :param selectable: any :class:`_expression.FromClause` subclass,
+        such as a table, select statement, etc.
+
+    :param name: string name to be assigned as the alias.
+        If ``None``, a name will be deterministically generated at compile
+        time. Deterministic means the name is guaranteed to be unique against
+        other constructs used in the same statement, and will also be the same
+        name for each successive compilation of the same statement object.
+
+    :param flat: Will be passed through to if the given selectable
+     is an instance of :class:`_expression.Join` - see
+     :meth:`_expression.Join.alias` for details.
+
+    """
+    return Alias._factory(selectable, name=name, flat=flat)
+
+
+def cte(
+    selectable: HasCTE, name: Optional[str] = None, recursive: bool = False
+) -> CTE:
+    r"""Return a new :class:`_expression.CTE`,
+    or Common Table Expression instance.
+
+    Please see :meth:`_expression.HasCTE.cte` for detail on CTE usage.
+
+    """
+    return coercions.expect(roles.HasCTERole, selectable).cte(
+        name=name, recursive=recursive
+    )
+
+
+# TODO: mypy requires the _TypedSelectable overloads in all compound select
+# constructors since _SelectStatementForCompoundArgument includes
+# untyped args that make it return CompoundSelect[Unpack[tuple[Never, ...]]]
+# pyright does not have this issue
+_TypedSelectable = Union["Select[_TP]", "CompoundSelect[_TP]"]
+
+
+@overload
+def except_(
+    *selects: _TypedSelectable[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+@overload
+def except_(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+def except_(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]:
+    r"""Return an ``EXCEPT`` of multiple selectables.
+
+    The returned object is an instance of
+    :class:`_expression.CompoundSelect`.
+
+    :param \*selects:
+      a list of :class:`_expression.Select` instances.
+
+    """
+    return CompoundSelect._create_except(*selects)
+
+
+@overload
+def except_all(
+    *selects: _TypedSelectable[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+@overload
+def except_all(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+def except_all(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]:
+    r"""Return an ``EXCEPT ALL`` of multiple selectables.
+
+    The returned object is an instance of
+    :class:`_expression.CompoundSelect`.
+
+    :param \*selects:
+      a list of :class:`_expression.Select` instances.
+
+    """
+    return CompoundSelect._create_except_all(*selects)
+
+
+def exists(
+    __argument: Optional[
+        Union[_ColumnsClauseArgument[Any], SelectBase, ScalarSelect[Any]]
+    ] = None,
+) -> Exists:
+    """Construct a new :class:`_expression.Exists` construct.
+
+    The :func:`_sql.exists` can be invoked by itself to produce an
+    :class:`_sql.Exists` construct, which will accept simple WHERE
+    criteria::
+
+        exists_criteria = exists().where(table1.c.col1 == table2.c.col2)
+
+    However, for greater flexibility in constructing the SELECT, an
+    existing :class:`_sql.Select` construct may be converted to an
+    :class:`_sql.Exists`, most conveniently by making use of the
+    :meth:`_sql.SelectBase.exists` method::
+
+        exists_criteria = (
+            select(table2.c.col2).where(table1.c.col1 == table2.c.col2).exists()
+        )
+
+    The EXISTS criteria is then used inside of an enclosing SELECT::
+
+        stmt = select(table1.c.col1).where(exists_criteria)
+
+    The above statement will then be of the form:
+
+    .. sourcecode:: sql
+
+        SELECT col1 FROM table1 WHERE EXISTS
+        (SELECT table2.col2 FROM table2 WHERE table2.col2 = table1.col1)
+
+    .. seealso::
+
+        :ref:`tutorial_exists` - in the :term:`2.0 style` tutorial.
+
+        :meth:`_sql.SelectBase.exists` - method to transform a ``SELECT`` to an
+        ``EXISTS`` clause.
+
+    """  # noqa: E501
+
+    return Exists(__argument)
+
+
+@overload
+def intersect(
+    *selects: _TypedSelectable[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+@overload
+def intersect(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+def intersect(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]:
+    r"""Return an ``INTERSECT`` of multiple selectables.
+
+    The returned object is an instance of
+    :class:`_expression.CompoundSelect`.
+
+    :param \*selects:
+      a list of :class:`_expression.Select` instances.
+
+    """
+    return CompoundSelect._create_intersect(*selects)
+
+
+@overload
+def intersect_all(
+    *selects: _TypedSelectable[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+@overload
+def intersect_all(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+def intersect_all(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]:
+    r"""Return an ``INTERSECT ALL`` of multiple selectables.
+
+    The returned object is an instance of
+    :class:`_expression.CompoundSelect`.
+
+    :param \*selects:
+      a list of :class:`_expression.Select` instances.
+
+
+    """
+    return CompoundSelect._create_intersect_all(*selects)
+
+
+def join(
+    left: _FromClauseArgument,
+    right: _FromClauseArgument,
+    onclause: Optional[_OnClauseArgument] = None,
+    isouter: bool = False,
+    full: bool = False,
+) -> Join:
+    """Produce a :class:`_expression.Join` object, given two
+    :class:`_expression.FromClause`
+    expressions.
+
+    E.g.::
+
+        j = join(
+            user_table, address_table, user_table.c.id == address_table.c.user_id
+        )
+        stmt = select(user_table).select_from(j)
+
+    would emit SQL along the lines of:
+
+    .. sourcecode:: sql
+
+        SELECT user.id, user.name FROM user
+        JOIN address ON user.id = address.user_id
+
+    Similar functionality is available given any
+    :class:`_expression.FromClause` object (e.g. such as a
+    :class:`_schema.Table`) using
+    the :meth:`_expression.FromClause.join` method.
+
+    :param left: The left side of the join.
+
+    :param right: the right side of the join; this is any
+     :class:`_expression.FromClause` object such as a
+     :class:`_schema.Table` object, and
+     may also be a selectable-compatible object such as an ORM-mapped
+     class.
+
+    :param onclause: a SQL expression representing the ON clause of the
+     join.  If left at ``None``, :meth:`_expression.FromClause.join`
+     will attempt to
+     join the two tables based on a foreign key relationship.
+
+    :param isouter: if True, render a LEFT OUTER JOIN, instead of JOIN.
+
+    :param full: if True, render a FULL OUTER JOIN, instead of JOIN.
+
+    .. seealso::
+
+        :meth:`_expression.FromClause.join` - method form,
+        based on a given left side.
+
+        :class:`_expression.Join` - the type of object produced.
+
+    """  # noqa: E501
+
+    return Join(left, right, onclause, isouter, full)
+
+
+def lateral(
+    selectable: Union[SelectBase, _FromClauseArgument],
+    name: Optional[str] = None,
+) -> LateralFromClause:
+    """Return a :class:`_expression.Lateral` object.
+
+    :class:`_expression.Lateral` is an :class:`_expression.Alias`
+    subclass that represents
+    a subquery with the LATERAL keyword applied to it.
+
+    The special behavior of a LATERAL subquery is that it appears in the
+    FROM clause of an enclosing SELECT, but may correlate to other
+    FROM clauses of that SELECT.   It is a special case of subquery
+    only supported by a small number of backends, currently more recent
+    PostgreSQL versions.
+
+    .. seealso::
+
+        :ref:`tutorial_lateral_correlation` -  overview of usage.
+
+    """
+    return Lateral._factory(selectable, name=name)
+
+
+def outerjoin(
+    left: _FromClauseArgument,
+    right: _FromClauseArgument,
+    onclause: Optional[_OnClauseArgument] = None,
+    full: bool = False,
+) -> Join:
+    """Return an ``OUTER JOIN`` clause element.
+
+    The returned object is an instance of :class:`_expression.Join`.
+
+    Similar functionality is also available via the
+    :meth:`_expression.FromClause.outerjoin` method on any
+    :class:`_expression.FromClause`.
+
+    :param left: The left side of the join.
+
+    :param right: The right side of the join.
+
+    :param onclause:  Optional criterion for the ``ON`` clause, is
+      derived from foreign key relationships established between
+      left and right otherwise.
+
+    To chain joins together, use the :meth:`_expression.FromClause.join`
+    or
+    :meth:`_expression.FromClause.outerjoin` methods on the resulting
+    :class:`_expression.Join` object.
+
+    """
+    return Join(left, right, onclause, isouter=True, full=full)
+
+
+# START OVERLOADED FUNCTIONS select Select 1-10
+
+# code within this block is **programmatically,
+# statically generated** by tools/generate_tuple_map_overloads.py
+
+
+@overload
+def select(__ent0: _TCCA[_T0]) -> Select[Tuple[_T0]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0], __ent1: _TCCA[_T1]
+) -> Select[Tuple[_T0, _T1]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
+) -> Select[Tuple[_T0, _T1, _T2]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0],
+    __ent1: _TCCA[_T1],
+    __ent2: _TCCA[_T2],
+    __ent3: _TCCA[_T3],
+) -> Select[Tuple[_T0, _T1, _T2, _T3]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0],
+    __ent1: _TCCA[_T1],
+    __ent2: _TCCA[_T2],
+    __ent3: _TCCA[_T3],
+    __ent4: _TCCA[_T4],
+) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0],
+    __ent1: _TCCA[_T1],
+    __ent2: _TCCA[_T2],
+    __ent3: _TCCA[_T3],
+    __ent4: _TCCA[_T4],
+    __ent5: _TCCA[_T5],
+) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0],
+    __ent1: _TCCA[_T1],
+    __ent2: _TCCA[_T2],
+    __ent3: _TCCA[_T3],
+    __ent4: _TCCA[_T4],
+    __ent5: _TCCA[_T5],
+    __ent6: _TCCA[_T6],
+) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0],
+    __ent1: _TCCA[_T1],
+    __ent2: _TCCA[_T2],
+    __ent3: _TCCA[_T3],
+    __ent4: _TCCA[_T4],
+    __ent5: _TCCA[_T5],
+    __ent6: _TCCA[_T6],
+    __ent7: _TCCA[_T7],
+) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0],
+    __ent1: _TCCA[_T1],
+    __ent2: _TCCA[_T2],
+    __ent3: _TCCA[_T3],
+    __ent4: _TCCA[_T4],
+    __ent5: _TCCA[_T5],
+    __ent6: _TCCA[_T6],
+    __ent7: _TCCA[_T7],
+    __ent8: _TCCA[_T8],
+) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7, _T8]]: ...
+
+
+@overload
+def select(
+    __ent0: _TCCA[_T0],
+    __ent1: _TCCA[_T1],
+    __ent2: _TCCA[_T2],
+    __ent3: _TCCA[_T3],
+    __ent4: _TCCA[_T4],
+    __ent5: _TCCA[_T5],
+    __ent6: _TCCA[_T6],
+    __ent7: _TCCA[_T7],
+    __ent8: _TCCA[_T8],
+    __ent9: _TCCA[_T9],
+) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7, _T8, _T9]]: ...
+
+
+# END OVERLOADED FUNCTIONS select
+
+
+@overload
+def select(
+    *entities: _ColumnsClauseArgument[Any], **__kw: Any
+) -> Select[Any]: ...
+
+
+def select(*entities: _ColumnsClauseArgument[Any], **__kw: Any) -> Select[Any]:
+    r"""Construct a new :class:`_expression.Select`.
+
+
+    .. versionadded:: 1.4 - The :func:`_sql.select` function now accepts
+       column arguments positionally.   The top-level :func:`_sql.select`
+       function will automatically use the 1.x or 2.x style API based on
+       the incoming arguments; using :func:`_sql.select` from the
+       ``sqlalchemy.future`` module will enforce that only the 2.x style
+       constructor is used.
+
+    Similar functionality is also available via the
+    :meth:`_expression.FromClause.select` method on any
+    :class:`_expression.FromClause`.
+
+    .. seealso::
+
+        :ref:`tutorial_selecting_data` - in the :ref:`unified_tutorial`
+
+    :param \*entities:
+      Entities to SELECT from.  For Core usage, this is typically a series
+      of :class:`_expression.ColumnElement` and / or
+      :class:`_expression.FromClause`
+      objects which will form the columns clause of the resulting
+      statement.   For those objects that are instances of
+      :class:`_expression.FromClause` (typically :class:`_schema.Table`
+      or :class:`_expression.Alias`
+      objects), the :attr:`_expression.FromClause.c`
+      collection is extracted
+      to form a collection of :class:`_expression.ColumnElement` objects.
+
+      This parameter will also accept :class:`_expression.TextClause`
+      constructs as
+      given, as well as ORM-mapped classes.
+
+    """
+    # the keyword args are a necessary element in order for the typing
+    # to work out w/ the varargs vs. having named "keyword" arguments that
+    # aren't always present.
+    if __kw:
+        raise _no_kw()
+    return Select(*entities)
+
+
+def table(name: str, *columns: ColumnClause[Any], **kw: Any) -> TableClause:
+    """Produce a new :class:`_expression.TableClause`.
+
+    The object returned is an instance of
+    :class:`_expression.TableClause`, which
+    represents the "syntactical" portion of the schema-level
+    :class:`_schema.Table` object.
+    It may be used to construct lightweight table constructs.
+
+    :param name: Name of the table.
+
+    :param columns: A collection of :func:`_expression.column` constructs.
+
+    :param schema: The schema name for this table.
+
+        .. versionadded:: 1.3.18 :func:`_expression.table` can now
+           accept a ``schema`` argument.
+    """
+
+    return TableClause(name, *columns, **kw)
+
+
+def tablesample(
+    selectable: _FromClauseArgument,
+    sampling: Union[float, Function[Any]],
+    name: Optional[str] = None,
+    seed: Optional[roles.ExpressionElementRole[Any]] = None,
+) -> TableSample:
+    """Return a :class:`_expression.TableSample` object.
+
+    :class:`_expression.TableSample` is an :class:`_expression.Alias`
+    subclass that represents
+    a table with the TABLESAMPLE clause applied to it.
+    :func:`_expression.tablesample`
+    is also available from the :class:`_expression.FromClause`
+    class via the
+    :meth:`_expression.FromClause.tablesample` method.
+
+    The TABLESAMPLE clause allows selecting a randomly selected approximate
+    percentage of rows from a table. It supports multiple sampling methods,
+    most commonly BERNOULLI and SYSTEM.
+
+    e.g.::
+
+        from sqlalchemy import func
+
+        selectable = people.tablesample(
+            func.bernoulli(1), name="alias", seed=func.random()
+        )
+        stmt = select(selectable.c.people_id)
+
+    Assuming ``people`` with a column ``people_id``, the above
+    statement would render as:
+
+    .. sourcecode:: sql
+
+        SELECT alias.people_id FROM
+        people AS alias TABLESAMPLE bernoulli(:bernoulli_1)
+        REPEATABLE (random())
+
+    :param sampling: a ``float`` percentage between 0 and 100 or
+        :class:`_functions.Function`.
+
+    :param name: optional alias name
+
+    :param seed: any real-valued SQL expression.  When specified, the
+     REPEATABLE sub-clause is also rendered.
+
+    """
+    return TableSample._factory(selectable, sampling, name=name, seed=seed)
+
+
+@overload
+def union(
+    *selects: _TypedSelectable[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+@overload
+def union(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+def union(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]:
+    r"""Return a ``UNION`` of multiple selectables.
+
+    The returned object is an instance of
+    :class:`_expression.CompoundSelect`.
+
+    A similar :func:`union()` method is available on all
+    :class:`_expression.FromClause` subclasses.
+
+    :param \*selects:
+      a list of :class:`_expression.Select` instances.
+
+    :param \**kwargs:
+      available keyword arguments are the same as those of
+      :func:`select`.
+
+    """
+    return CompoundSelect._create_union(*selects)
+
+
+@overload
+def union_all(
+    *selects: _TypedSelectable[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+@overload
+def union_all(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]: ...
+
+
+def union_all(
+    *selects: _SelectStatementForCompoundArgument[_TP],
+) -> CompoundSelect[_TP]:
+    r"""Return a ``UNION ALL`` of multiple selectables.
+
+    The returned object is an instance of
+    :class:`_expression.CompoundSelect`.
+
+    A similar :func:`union_all()` method is available on all
+    :class:`_expression.FromClause` subclasses.
+
+    :param \*selects:
+      a list of :class:`_expression.Select` instances.
+
+    """
+    return CompoundSelect._create_union_all(*selects)
+
+
+def values(
+    *columns: _OnlyColumnArgument[Any],
+    name: Optional[str] = None,
+    literal_binds: bool = False,
+) -> Values:
+    r"""Construct a :class:`_expression.Values` construct representing the
+    SQL ``VALUES`` clause.
+
+    The column expressions and the actual data for :class:`_expression.Values`
+    are given in two separate steps.  The constructor receives the column
+    expressions typically as :func:`_expression.column` constructs, and the
+    data is then passed via the :meth:`_expression.Values.data` method as a
+    list, which can be called multiple times to add more data, e.g.::
+
+        from sqlalchemy import column
+        from sqlalchemy import values
+        from sqlalchemy import Integer
+        from sqlalchemy import String
+
+        value_expr = (
+            values(
+                column("id", Integer),
+                column("name", String),
+            )
+            .data([(1, "name1"), (2, "name2")])
+            .data([(3, "name3")])
+        )
+
+    Would represent a SQL fragment like::
+
+        VALUES(1, "name1"), (2, "name2"), (3, "name3")
+
+    The :class:`_sql.values` construct has an optional
+    :paramref:`_sql.values.name` field; when using this field, the
+    PostgreSQL-specific "named VALUES" clause may be generated::
+
+        value_expr = values(
+            column("id", Integer), column("name", String), name="somename"
+        ).data([(1, "name1"), (2, "name2"), (3, "name3")])
+
+    When selecting from the above construct, the name and column names will
+    be listed out using a PostgreSQL-specific syntax::
+
+        >>> print(value_expr.select())
+        SELECT somename.id, somename.name
+        FROM (VALUES (:param_1, :param_2), (:param_3, :param_4),
+        (:param_5, :param_6)) AS somename (id, name)
+
+    For a more database-agnostic means of SELECTing named columns from a
+    VALUES expression, the :meth:`.Values.cte` method may be used, which
+    produces a named CTE with explicit column names against the VALUES
+    construct within; this syntax works on PostgreSQL, SQLite, and MariaDB::
+
+        value_expr = (
+            values(
+                column("id", Integer),
+                column("name", String),
+            )
+            .data([(1, "name1"), (2, "name2"), (3, "name3")])
+            .cte()
+        )
+
+    Rendering as::
+
+        >>> print(value_expr.select())
+        WITH anon_1(id, name) AS
+        (VALUES (:param_1, :param_2), (:param_3, :param_4), (:param_5, :param_6))
+        SELECT anon_1.id, anon_1.name
+        FROM anon_1
+
+    .. versionadded:: 2.0.42  Added the :meth:`.Values.cte` method to
+       :class:`.Values`
+
+    :param \*columns: column expressions, typically composed using
+     :func:`_expression.column` objects.
+
+    :param name: the name for this VALUES construct.  If omitted, the
+     VALUES construct will be unnamed in a SQL expression.   Different
+     backends may have different requirements here.
+
+    :param literal_binds: Defaults to False.  Whether or not to render
+     the data values inline in the SQL output, rather than using bound
+     parameters.
+
+    """  # noqa: E501
+
+    return Values(*columns, literal_binds=literal_binds, name=name)

venv/Lib/site-packages/sqlalchemy/sql/_typing.py

@@ -0,0 +1,482 @@
+# sql/_typing.py
+# Copyright (C) 2022-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
+
+from __future__ import annotations
+
+import operator
+from typing import Any
+from typing import Callable
+from typing import Dict
+from typing import Generic
+from typing import Iterable
+from typing import Mapping
+from typing import NoReturn
+from typing import Optional
+from typing import overload
+from typing import Set
+from typing import Tuple
+from typing import Type
+from typing import TYPE_CHECKING
+from typing import TypeVar
+from typing import Union
+
+from . import roles
+from .. import exc
+from .. import util
+from ..inspection import Inspectable
+from ..util.typing import Literal
+from ..util.typing import Protocol
+from ..util.typing import TypeAlias
+
+if TYPE_CHECKING:
+    from datetime import date
+    from datetime import datetime
+    from datetime import time
+    from datetime import timedelta
+    from decimal import Decimal
+    from uuid import UUID
+
+    from .base import Executable
+    from .compiler import Compiled
+    from .compiler import DDLCompiler
+    from .compiler import SQLCompiler
+    from .dml import UpdateBase
+    from .dml import ValuesBase
+    from .elements import ClauseElement
+    from .elements import ColumnElement
+    from .elements import KeyedColumnElement
+    from .elements import quoted_name
+    from .elements import SQLCoreOperations
+    from .elements import TextClause
+    from .lambdas import LambdaElement
+    from .roles import FromClauseRole
+    from .schema import Column
+    from .selectable import Alias
+    from .selectable import CompoundSelect
+    from .selectable import CTE
+    from .selectable import FromClause
+    from .selectable import Join
+    from .selectable import NamedFromClause
+    from .selectable import ReturnsRows
+    from .selectable import Select
+    from .selectable import Selectable
+    from .selectable import SelectBase
+    from .selectable import Subquery
+    from .selectable import TableClause
+    from .sqltypes import TableValueType
+    from .sqltypes import TupleType
+    from .type_api import TypeEngine
+    from ..engine import Connection
+    from ..engine import Dialect
+    from ..engine import Engine
+    from ..engine.mock import MockConnection
+    from ..util.typing import TypeGuard
+
+_T = TypeVar("_T", bound=Any)
+_T_co = TypeVar("_T_co", bound=Any, covariant=True)
+
+
+_CE = TypeVar("_CE", bound="ColumnElement[Any]")
+
+_CLE = TypeVar("_CLE", bound="ClauseElement")
+
+
+class _HasClauseElement(Protocol, Generic[_T_co]):
+    """indicates a class that has a __clause_element__() method"""
+
+    def __clause_element__(self) -> roles.ExpressionElementRole[_T_co]: ...
+
+
+class _CoreAdapterProto(Protocol):
+    """protocol for the ClauseAdapter/ColumnAdapter.traverse() method."""
+
+    def __call__(self, obj: _CE) -> _CE: ...
+
+
+class _HasDialect(Protocol):
+    """protocol for Engine/Connection-like objects that have dialect
+    attribute.
+    """
+
+    @property
+    def dialect(self) -> Dialect: ...
+
+
+# match column types that are not ORM entities
+_NOT_ENTITY = TypeVar(
+    "_NOT_ENTITY",
+    int,
+    str,
+    bool,
+    "datetime",
+    "date",
+    "time",
+    "timedelta",
+    "UUID",
+    float,
+    "Decimal",
+)
+
+_StarOrOne = Literal["*", 1]
+
+_MAYBE_ENTITY = TypeVar(
+    "_MAYBE_ENTITY",
+    roles.ColumnsClauseRole,
+    _StarOrOne,
+    Type[Any],
+    Inspectable[_HasClauseElement[Any]],
+    _HasClauseElement[Any],
+)
+
+
+# convention:
+# XYZArgument - something that the end user is passing to a public API method
+# XYZElement - the internal representation that we use for the thing.
+# the coercions system is responsible for converting from XYZArgument to
+# XYZElement.
+
+_TextCoercedExpressionArgument = Union[
+    str,
+    "TextClause",
+    "ColumnElement[_T]",
+    _HasClauseElement[_T],
+    roles.ExpressionElementRole[_T],
+]
+
+_ColumnsClauseArgument = Union[
+    roles.TypedColumnsClauseRole[_T],
+    roles.ColumnsClauseRole,
+    "SQLCoreOperations[_T]",
+    _StarOrOne,
+    Type[_T],
+    Inspectable[_HasClauseElement[_T]],
+    _HasClauseElement[_T],
+]
+"""open-ended SELECT columns clause argument.
+
+Includes column expressions, tables, ORM mapped entities, a few literal values.
+
+This type is used for lists of columns  / entities to be returned in result
+sets; select(...), insert().returning(...), etc.
+
+
+"""
+
+_TypedColumnClauseArgument = Union[
+    roles.TypedColumnsClauseRole[_T],
+    "SQLCoreOperations[_T]",
+    Type[_T],
+]
+
+_TP = TypeVar("_TP", bound=Tuple[Any, ...])
+
+_T0 = TypeVar("_T0", bound=Any)
+_T1 = TypeVar("_T1", bound=Any)
+_T2 = TypeVar("_T2", bound=Any)
+_T3 = TypeVar("_T3", bound=Any)
+_T4 = TypeVar("_T4", bound=Any)
+_T5 = TypeVar("_T5", bound=Any)
+_T6 = TypeVar("_T6", bound=Any)
+_T7 = TypeVar("_T7", bound=Any)
+_T8 = TypeVar("_T8", bound=Any)
+_T9 = TypeVar("_T9", bound=Any)
+
+
+_OnlyColumnArgument = Union[
+    "ColumnElement[_T]",
+    _HasClauseElement[_T],
+    roles.DMLColumnRole,
+]
+"""A narrow type that is looking for a ColumnClause (e.g. table column with a
+name) or an ORM element that produces this.
+
+This is used for constructs that need a named column to represent a
+position in a selectable, like TextClause().columns() or values(...).
+
+"""
+
+_ColumnExpressionArgument = Union[
+    "ColumnElement[_T]",
+    _HasClauseElement[_T],
+    "SQLCoreOperations[_T]",
+    roles.ExpressionElementRole[_T],
+    roles.TypedColumnsClauseRole[_T],
+    Callable[[], "ColumnElement[_T]"],
+    "LambdaElement",
+]
+"See docs in public alias ColumnExpressionArgument."
+
+ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T]
+"""Narrower "column expression" argument.
+
+This type is used for all the other "column" kinds of expressions that
+typically represent a single SQL column expression, not a set of columns the
+way a table or ORM entity does.
+
+This includes ColumnElement, or ORM-mapped attributes that will have a
+``__clause_element__()`` method, it also has the ExpressionElementRole
+overall which brings in the TextClause object also.
+
+.. versionadded:: 2.0.13
+
+"""
+
+_ColumnExpressionOrLiteralArgument = Union[Any, _ColumnExpressionArgument[_T]]
+
+_ColumnExpressionOrStrLabelArgument = Union[str, _ColumnExpressionArgument[_T]]
+
+_ByArgument = Union[
+    Iterable[_ColumnExpressionOrStrLabelArgument[Any]],
+    _ColumnExpressionOrStrLabelArgument[Any],
+]
+"""Used for keyword-based ``order_by`` and ``partition_by`` parameters."""
+
+
+_InfoType = Dict[Any, Any]
+"""the .info dictionary accepted and used throughout Core /ORM"""
+
+_FromClauseArgument = Union[
+    roles.FromClauseRole,
+    roles.TypedColumnsClauseRole[Any],
+    Type[Any],
+    Inspectable[_HasClauseElement[Any]],
+    _HasClauseElement[Any],
+]
+"""A FROM clause, like we would send to select().select_from().
+
+Also accommodates ORM entities and related constructs.
+
+"""
+
+_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole]
+"""target for join() builds on _FromClauseArgument to include additional
+join target roles such as those which come from the ORM.
+
+"""
+
+_OnClauseArgument = Union[_ColumnExpressionArgument[Any], roles.OnClauseRole]
+"""target for an ON clause, includes additional roles such as those which
+come from the ORM.
+
+"""
+
+_SelectStatementForCompoundArgument = Union[
+    "Select[_TP]",
+    "CompoundSelect[_TP]",
+    roles.CompoundElementRole,
+]
+"""SELECT statement acceptable by ``union()`` and other SQL set operations"""
+
+_DMLColumnArgument = Union[
+    str,
+    _HasClauseElement[Any],
+    roles.DMLColumnRole,
+    "SQLCoreOperations[Any]",
+]
+"""A DML column expression.  This is a "key" inside of insert().values(),
+update().values(), and related.
+
+These are usually strings or SQL table columns.
+
+There's also edge cases like JSON expression assignment, which we would want
+the DMLColumnRole to be able to accommodate.
+
+"""
+
+_DMLKey = TypeVar("_DMLKey", bound=_DMLColumnArgument)
+_DMLColumnKeyMapping = Mapping[_DMLKey, Any]
+
+
+_DDLColumnArgument = Union[str, "Column[Any]", roles.DDLConstraintColumnRole]
+"""DDL column.
+
+used for :class:`.PrimaryKeyConstraint`, :class:`.UniqueConstraint`, etc.
+
+"""
+
+_DMLTableArgument = Union[
+    "TableClause",
+    "Join",
+    "Alias",
+    "CTE",
+    Type[Any],
+    Inspectable[_HasClauseElement[Any]],
+    _HasClauseElement[Any],
+]
+
+_PropagateAttrsType = util.immutabledict[str, Any]
+
+_TypeEngineArgument = Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"]
+
+_EquivalentColumnMap = Dict["ColumnElement[Any]", Set["ColumnElement[Any]"]]
+
+_LimitOffsetType = Union[int, _ColumnExpressionArgument[int], None]
+
+_AutoIncrementType = Union[bool, Literal["auto", "ignore_fk"]]
+
+_CreateDropBind = Union["Engine", "Connection", "MockConnection"]
+
+if TYPE_CHECKING:
+
+    def is_sql_compiler(c: Compiled) -> TypeGuard[SQLCompiler]: ...
+
+    def is_ddl_compiler(c: Compiled) -> TypeGuard[DDLCompiler]: ...
+
+    def is_named_from_clause(
+        t: FromClauseRole,
+    ) -> TypeGuard[NamedFromClause]: ...
+
+    def is_column_element(
+        c: ClauseElement,
+    ) -> TypeGuard[ColumnElement[Any]]: ...
+
+    def is_keyed_column_element(
+        c: ClauseElement,
+    ) -> TypeGuard[KeyedColumnElement[Any]]: ...
+
+    def is_text_clause(c: ClauseElement) -> TypeGuard[TextClause]: ...
+
+    def is_from_clause(c: ClauseElement) -> TypeGuard[FromClause]: ...
+
+    def is_tuple_type(t: TypeEngine[Any]) -> TypeGuard[TupleType]: ...
+
+    def is_table_value_type(
+        t: TypeEngine[Any],
+    ) -> TypeGuard[TableValueType]: ...
+
+    def is_selectable(t: Any) -> TypeGuard[Selectable]: ...
+
+    def is_select_base(
+        t: Union[Executable, ReturnsRows],
+    ) -> TypeGuard[SelectBase]: ...
+
+    def is_select_statement(
+        t: Union[Executable, ReturnsRows],
+    ) -> TypeGuard[Select[Any]]: ...
+
+    def is_table(t: FromClause) -> TypeGuard[TableClause]: ...
+
+    def is_subquery(t: FromClause) -> TypeGuard[Subquery]: ...
+
+    def is_dml(c: ClauseElement) -> TypeGuard[UpdateBase]: ...
+
+else:
+    is_sql_compiler = operator.attrgetter("is_sql")
+    is_ddl_compiler = operator.attrgetter("is_ddl")
+    is_named_from_clause = operator.attrgetter("named_with_column")
+    is_column_element = operator.attrgetter("_is_column_element")
+    is_keyed_column_element = operator.attrgetter("_is_keyed_column_element")
+    is_text_clause = operator.attrgetter("_is_text_clause")
+    is_from_clause = operator.attrgetter("_is_from_clause")
+    is_tuple_type = operator.attrgetter("_is_tuple_type")
+    is_table_value_type = operator.attrgetter("_is_table_value")
+    is_selectable = operator.attrgetter("is_selectable")
+    is_select_base = operator.attrgetter("_is_select_base")
+    is_select_statement = operator.attrgetter("_is_select_statement")
+    is_table = operator.attrgetter("_is_table")
+    is_subquery = operator.attrgetter("_is_subquery")
+    is_dml = operator.attrgetter("is_dml")
+
+
+def has_schema_attr(t: FromClauseRole) -> TypeGuard[TableClause]:
+    return hasattr(t, "schema")
+
+
+def is_quoted_name(s: str) -> TypeGuard[quoted_name]:
+    return hasattr(s, "quote")
+
+
+def is_has_clause_element(s: object) -> TypeGuard[_HasClauseElement[Any]]:
+    return hasattr(s, "__clause_element__")
+
+
+def is_insert_update(c: ClauseElement) -> TypeGuard[ValuesBase]:
+    return c.is_dml and (c.is_insert or c.is_update)  # type: ignore
+
+
+def _no_kw() -> exc.ArgumentError:
+    return exc.ArgumentError(
+        "Additional keyword arguments are not accepted by this "
+        "function/method.  The presence of **kw is for pep-484 typing purposes"
+    )
+
+
+def _unexpected_kw(methname: str, kw: Dict[str, Any]) -> NoReturn:
+    k = list(kw)[0]
+    raise TypeError(f"{methname} got an unexpected keyword argument '{k}'")
+
+
+@overload
+def Nullable(
+    val: "SQLCoreOperations[_T]",
+) -> "SQLCoreOperations[Optional[_T]]": ...
+
+
+@overload
+def Nullable(
+    val: roles.ExpressionElementRole[_T],
+) -> roles.ExpressionElementRole[Optional[_T]]: ...
+
+
+@overload
+def Nullable(val: Type[_T]) -> Type[Optional[_T]]: ...
+
+
+def Nullable(
+    val: _TypedColumnClauseArgument[_T],
+) -> _TypedColumnClauseArgument[Optional[_T]]:
+    """Types a column or ORM class as nullable.
+
+    This can be used in select and other contexts to express that the value of
+    a column can be null, for example due to an outer join::
+
+        stmt1 = select(A, Nullable(B)).outerjoin(A.bs)
+        stmt2 = select(A.data, Nullable(B.data)).outerjoin(A.bs)
+
+    At runtime this method returns the input unchanged.
+
+    .. versionadded:: 2.0.20
+    """
+    return val
+
+
+@overload
+def NotNullable(
+    val: "SQLCoreOperations[Optional[_T]]",
+) -> "SQLCoreOperations[_T]": ...
+
+
+@overload
+def NotNullable(
+    val: roles.ExpressionElementRole[Optional[_T]],
+) -> roles.ExpressionElementRole[_T]: ...
+
+
+@overload
+def NotNullable(val: Type[Optional[_T]]) -> Type[_T]: ...
+
+
+@overload
+def NotNullable(val: Optional[Type[_T]]) -> Type[_T]: ...
+
+
+def NotNullable(
+    val: Union[_TypedColumnClauseArgument[Optional[_T]], Optional[Type[_T]]],
+) -> _TypedColumnClauseArgument[_T]:
+    """Types a column or ORM class as not nullable.
+
+    This can be used in select and other contexts to express that the value of
+    a column cannot be null, for example due to a where condition on a
+    nullable column::
+
+        stmt = select(NotNullable(A.value)).where(A.value.is_not(None))
+
+    At runtime this method returns the input unchanged.
+
+    .. versionadded:: 2.0.20
+    """
+    return val  # type: ignore

venv/Lib/site-packages/sqlalchemy/sql/annotation.py

@@ -0,0 +1,587 @@
+# sql/annotation.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
+
+"""The :class:`.Annotated` class and related routines; creates hash-equivalent
+copies of SQL constructs which contain context-specific markers and
+associations.
+
+Note that the :class:`.Annotated` concept as implemented in this module is not
+related in any way to the pep-593 concept of "Annotated".
+
+
+"""
+
+from __future__ import annotations
+
+import typing
+from typing import Any
+from typing import Callable
+from typing import cast
+from typing import Dict
+from typing import FrozenSet
+from typing import Mapping
+from typing import Optional
+from typing import overload
+from typing import Sequence
+from typing import Tuple
+from typing import Type
+from typing import TYPE_CHECKING
+from typing import TypeVar
+
+from . import operators
+from .cache_key import HasCacheKey
+from .visitors import anon_map
+from .visitors import ExternallyTraversible
+from .visitors import InternalTraversal
+from .. import util
+from ..util.typing import Literal
+from ..util.typing import Self
+
+if TYPE_CHECKING:
+    from .base import _EntityNamespace
+    from .visitors import _TraverseInternalsType
+
+_AnnotationDict = Mapping[str, Any]
+
+EMPTY_ANNOTATIONS: util.immutabledict[str, Any] = util.EMPTY_DICT
+
+
+class SupportsAnnotations(ExternallyTraversible):
+    __slots__ = ()
+
+    _annotations: util.immutabledict[str, Any] = EMPTY_ANNOTATIONS
+
+    proxy_set: util.generic_fn_descriptor[FrozenSet[Any]]
+
+    _is_immutable: bool
+
+    def _annotate(self, values: _AnnotationDict) -> Self:
+        raise NotImplementedError()
+
+    @overload
+    def _deannotate(
+        self,
+        values: Literal[None] = ...,
+        clone: bool = ...,
+    ) -> Self: ...
+
+    @overload
+    def _deannotate(
+        self,
+        values: Sequence[str] = ...,
+        clone: bool = ...,
+    ) -> SupportsAnnotations: ...
+
+    def _deannotate(
+        self,
+        values: Optional[Sequence[str]] = None,
+        clone: bool = False,
+    ) -> SupportsAnnotations:
+        raise NotImplementedError()
+
+    @util.memoized_property
+    def _annotations_cache_key(self) -> Tuple[Any, ...]:
+        anon_map_ = anon_map()
+
+        return self._gen_annotations_cache_key(anon_map_)
+
+    def _gen_annotations_cache_key(
+        self, anon_map: anon_map
+    ) -> Tuple[Any, ...]:
+        return (
+            "_annotations",
+            tuple(
+                (
+                    key,
+                    (
+                        value._gen_cache_key(anon_map, [])
+                        if isinstance(value, HasCacheKey)
+                        else value
+                    ),
+                )
+                for key, value in [
+                    (key, self._annotations[key])
+                    for key in sorted(self._annotations)
+                ]
+            ),
+        )
+
+
+class SupportsWrappingAnnotations(SupportsAnnotations):
+    __slots__ = ()
+
+    _constructor: Callable[..., SupportsWrappingAnnotations]
+
+    if TYPE_CHECKING:
+
+        @util.ro_non_memoized_property
+        def entity_namespace(self) -> _EntityNamespace: ...
+
+    def _annotate(self, values: _AnnotationDict) -> Self:
+        """return a copy of this ClauseElement with annotations
+        updated by the given dictionary.
+
+        """
+        return Annotated._as_annotated_instance(self, values)  # type: ignore
+
+    def _with_annotations(self, values: _AnnotationDict) -> Self:
+        """return a copy of this ClauseElement with annotations
+        replaced by the given dictionary.
+
+        """
+        return Annotated._as_annotated_instance(self, values)  # type: ignore
+
+    @overload
+    def _deannotate(
+        self,
+        values: Literal[None] = ...,
+        clone: bool = ...,
+    ) -> Self: ...
+
+    @overload
+    def _deannotate(
+        self,
+        values: Sequence[str] = ...,
+        clone: bool = ...,
+    ) -> SupportsAnnotations: ...
+
+    def _deannotate(
+        self,
+        values: Optional[Sequence[str]] = None,
+        clone: bool = False,
+    ) -> SupportsAnnotations:
+        """return a copy of this :class:`_expression.ClauseElement`
+        with annotations
+        removed.
+
+        :param values: optional tuple of individual values
+         to remove.
+
+        """
+        if clone:
+            s = self._clone()
+            return s
+        else:
+            return self
+
+
+class SupportsCloneAnnotations(SupportsWrappingAnnotations):
+    # SupportsCloneAnnotations extends from SupportsWrappingAnnotations
+    # to support the structure of having the base ClauseElement
+    # be a subclass of SupportsWrappingAnnotations.  Any ClauseElement
+    # subclass that wants to extend from SupportsCloneAnnotations
+    # will inherently also be subclassing SupportsWrappingAnnotations, so
+    # make that specific here.
+
+    if not typing.TYPE_CHECKING:
+        __slots__ = ()
+
+    _clone_annotations_traverse_internals: _TraverseInternalsType = [
+        ("_annotations", InternalTraversal.dp_annotations_key)
+    ]
+
+    def _annotate(self, values: _AnnotationDict) -> Self:
+        """return a copy of this ClauseElement with annotations
+        updated by the given dictionary.
+
+        """
+        new = self._clone()
+        new._annotations = new._annotations.union(values)
+        new.__dict__.pop("_annotations_cache_key", None)
+        new.__dict__.pop("_generate_cache_key", None)
+        return new
+
+    def _with_annotations(self, values: _AnnotationDict) -> Self:
+        """return a copy of this ClauseElement with annotations
+        replaced by the given dictionary.
+
+        """
+        new = self._clone()
+        new._annotations = util.immutabledict(values)
+        new.__dict__.pop("_annotations_cache_key", None)
+        new.__dict__.pop("_generate_cache_key", None)
+        return new
+
+    @overload
+    def _deannotate(
+        self,
+        values: Literal[None] = ...,
+        clone: bool = ...,
+    ) -> Self: ...
+
+    @overload
+    def _deannotate(
+        self,
+        values: Sequence[str] = ...,
+        clone: bool = ...,
+    ) -> SupportsAnnotations: ...
+
+    def _deannotate(
+        self,
+        values: Optional[Sequence[str]] = None,
+        clone: bool = False,
+    ) -> SupportsAnnotations:
+        """return a copy of this :class:`_expression.ClauseElement`
+        with annotations
+        removed.
+
+        :param values: optional tuple of individual values
+         to remove.
+
+        """
+        if clone or self._annotations:
+            # clone is used when we are also copying
+            # the expression for a deep deannotation
+            new = self._clone()
+            new._annotations = util.immutabledict()
+            new.__dict__.pop("_annotations_cache_key", None)
+            return new
+        else:
+            return self
+
+
+class Annotated(SupportsAnnotations):
+    """clones a SupportsAnnotations and applies an 'annotations' dictionary.
+
+    Unlike regular clones, this clone also mimics __hash__() and
+    __eq__() of the original element so that it takes its place
+    in hashed collections.
+
+    A reference to the original element is maintained, for the important
+    reason of keeping its hash value current.  When GC'ed, the
+    hash value may be reused, causing conflicts.
+
+    .. note::  The rationale for Annotated producing a brand new class,
+       rather than placing the functionality directly within ClauseElement,
+       is **performance**.  The __hash__() method is absent on plain
+       ClauseElement which leads to significantly reduced function call
+       overhead, as the use of sets and dictionaries against ClauseElement
+       objects is prevalent, but most are not "annotated".
+
+    """
+
+    _is_column_operators = False
+
+    @classmethod
+    def _as_annotated_instance(
+        cls, element: SupportsWrappingAnnotations, values: _AnnotationDict
+    ) -> Annotated:
+        try:
+            cls = annotated_classes[element.__class__]
+        except KeyError:
+            cls = _new_annotation_type(element.__class__, cls)
+        return cls(element, values)
+
+    _annotations: util.immutabledict[str, Any]
+    __element: SupportsWrappingAnnotations
+    _hash: int
+
+    def __new__(cls: Type[Self], *args: Any) -> Self:
+        return object.__new__(cls)
+
+    def __init__(
+        self, element: SupportsWrappingAnnotations, values: _AnnotationDict
+    ):
+        self.__dict__ = element.__dict__.copy()
+        self.__dict__.pop("_annotations_cache_key", None)
+        self.__dict__.pop("_generate_cache_key", None)
+        self.__element = element
+        self._annotations = util.immutabledict(values)
+        self._hash = hash(element)
+
+    def _annotate(self, values: _AnnotationDict) -> Self:
+        _values = self._annotations.union(values)
+        new = self._with_annotations(_values)
+        return new
+
+    def _with_annotations(self, values: _AnnotationDict) -> Self:
+        clone = self.__class__.__new__(self.__class__)
+        clone.__dict__ = self.__dict__.copy()
+        clone.__dict__.pop("_annotations_cache_key", None)
+        clone.__dict__.pop("_generate_cache_key", None)
+        clone._annotations = util.immutabledict(values)
+        return clone
+
+    @overload
+    def _deannotate(
+        self,
+        values: Literal[None] = ...,
+        clone: bool = ...,
+    ) -> Self: ...
+
+    @overload
+    def _deannotate(
+        self,
+        values: Sequence[str] = ...,
+        clone: bool = ...,
+    ) -> Annotated: ...
+
+    def _deannotate(
+        self,
+        values: Optional[Sequence[str]] = None,
+        clone: bool = True,
+    ) -> SupportsAnnotations:
+        if values is None:
+            return self.__element
+        else:
+            return self._with_annotations(
+                util.immutabledict(
+                    {
+                        key: value
+                        for key, value in self._annotations.items()
+                        if key not in values
+                    }
+                )
+            )
+
+    if not typing.TYPE_CHECKING:
+        # manually proxy some methods that need extra attention
+        def _compiler_dispatch(self, visitor: Any, **kw: Any) -> Any:
+            return self.__element.__class__._compiler_dispatch(
+                self, visitor, **kw
+            )
+
+        @property
+        def _constructor(self):
+            return self.__element._constructor
+
+    def _clone(self, **kw: Any) -> Self:
+        clone = self.__element._clone(**kw)
+        if clone is self.__element:
+            # detect immutable, don't change anything
+            return self
+        else:
+            # update the clone with any changes that have occurred
+            # to this object's __dict__.
+            clone.__dict__.update(self.__dict__)
+            return self.__class__(clone, self._annotations)
+
+    def __reduce__(self) -> Tuple[Type[Annotated], Tuple[Any, ...]]:
+        return self.__class__, (self.__element, self._annotations)
+
+    def __hash__(self) -> int:
+        return self._hash
+
+    def __eq__(self, other: Any) -> bool:
+        if self._is_column_operators:
+            return self.__element.__class__.__eq__(self, other)
+        else:
+            return hash(other) == hash(self)
+
+    @util.ro_non_memoized_property
+    def entity_namespace(self) -> _EntityNamespace:
+        if "entity_namespace" in self._annotations:
+            return cast(
+                SupportsWrappingAnnotations,
+                self._annotations["entity_namespace"],
+            ).entity_namespace
+        else:
+            return self.__element.entity_namespace
+
+
+# hard-generate Annotated subclasses.  this technique
+# is used instead of on-the-fly types (i.e. type.__new__())
+# so that the resulting objects are pickleable; additionally, other
+# decisions can be made up front about the type of object being annotated
+# just once per class rather than per-instance.
+annotated_classes: Dict[Type[SupportsWrappingAnnotations], Type[Annotated]] = (
+    {}
+)
+
+_SA = TypeVar("_SA", bound="SupportsAnnotations")
+
+
+def _safe_annotate(to_annotate: _SA, annotations: _AnnotationDict) -> _SA:
+    try:
+        _annotate = to_annotate._annotate
+    except AttributeError:
+        # skip objects that don't actually have an `_annotate`
+        # attribute, namely QueryableAttribute inside of a join
+        # condition
+        return to_annotate
+    else:
+        return _annotate(annotations)
+
+
+def _deep_annotate(
+    element: _SA,
+    annotations: _AnnotationDict,
+    exclude: Optional[Sequence[SupportsAnnotations]] = None,
+    *,
+    detect_subquery_cols: bool = False,
+    ind_cols_on_fromclause: bool = False,
+    annotate_callable: Optional[
+        Callable[[SupportsAnnotations, _AnnotationDict], SupportsAnnotations]
+    ] = None,
+) -> _SA:
+    """Deep copy the given ClauseElement, annotating each element
+    with the given annotations dictionary.
+
+    Elements within the exclude collection will be cloned but not annotated.
+
+    """
+
+    # annotated objects hack the __hash__() method so if we want to
+    # uniquely process them we have to use id()
+
+    cloned_ids: Dict[int, SupportsAnnotations] = {}
+
+    def clone(elem: SupportsAnnotations, **kw: Any) -> SupportsAnnotations:
+        # ind_cols_on_fromclause means make sure an AnnotatedFromClause
+        # has its own .c collection independent of that which its proxying.
+        # this is used specifically by orm.LoaderCriteriaOption to break
+        # a reference cycle that it's otherwise prone to building,
+        # see test_relationship_criteria->
+        # test_loader_criteria_subquery_w_same_entity.  logic here was
+        # changed for #8796 and made explicit; previously it occurred
+        # by accident
+
+        kw["detect_subquery_cols"] = detect_subquery_cols
+        id_ = id(elem)
+
+        if id_ in cloned_ids:
+            return cloned_ids[id_]
+
+        if (
+            exclude
+            and hasattr(elem, "proxy_set")
+            and elem.proxy_set.intersection(exclude)
+        ):
+            newelem = elem._clone(clone=clone, **kw)
+        elif annotations != elem._annotations:
+            if detect_subquery_cols and elem._is_immutable:
+                to_annotate = elem._clone(clone=clone, **kw)
+            else:
+                to_annotate = elem
+            if annotate_callable:
+                newelem = annotate_callable(to_annotate, annotations)
+            else:
+                newelem = _safe_annotate(to_annotate, annotations)
+        else:
+            newelem = elem
+
+        newelem._copy_internals(
+            clone=clone,
+            ind_cols_on_fromclause=ind_cols_on_fromclause,
+            _annotations_traversal=True,
+        )
+
+        cloned_ids[id_] = newelem
+        return newelem
+
+    if element is not None:
+        element = cast(_SA, clone(element))
+    clone = None  # type: ignore  # remove gc cycles
+    return element
+
+
+@overload
+def _deep_deannotate(
+    element: Literal[None], values: Optional[Sequence[str]] = None
+) -> Literal[None]: ...
+
+
+@overload
+def _deep_deannotate(
+    element: _SA, values: Optional[Sequence[str]] = None
+) -> _SA: ...
+
+
+def _deep_deannotate(
+    element: Optional[_SA], values: Optional[Sequence[str]] = None
+) -> Optional[_SA]:
+    """Deep copy the given element, removing annotations."""
+
+    cloned: Dict[Any, SupportsAnnotations] = {}
+
+    def clone(elem: SupportsAnnotations, **kw: Any) -> SupportsAnnotations:
+        key: Any
+        if values:
+            key = id(elem)
+        else:
+            key = elem
+
+        if key not in cloned:
+            newelem = elem._deannotate(values=values, clone=True)
+            newelem._copy_internals(clone=clone, _annotations_traversal=True)
+            cloned[key] = newelem
+            return newelem
+        else:
+            return cloned[key]
+
+    if element is not None:
+        element = cast(_SA, clone(element))
+    clone = None  # type: ignore  # remove gc cycles
+    return element
+
+
+def _shallow_annotate(element: _SA, annotations: _AnnotationDict) -> _SA:
+    """Annotate the given ClauseElement and copy its internals so that
+    internal objects refer to the new annotated object.
+
+    Basically used to apply a "don't traverse" annotation to a
+    selectable, without digging throughout the whole
+    structure wasting time.
+    """
+    element = element._annotate(annotations)
+    element._copy_internals(_annotations_traversal=True)
+    return element
+
+
+def _new_annotation_type(
+    cls: Type[SupportsWrappingAnnotations], base_cls: Type[Annotated]
+) -> Type[Annotated]:
+    """Generates a new class that subclasses Annotated and proxies a given
+    element type.
+
+    """
+    if issubclass(cls, Annotated):
+        return cls
+    elif cls in annotated_classes:
+        return annotated_classes[cls]
+
+    for super_ in cls.__mro__:
+        # check if an Annotated subclass more specific than
+        # the given base_cls is already registered, such
+        # as AnnotatedColumnElement.
+        if super_ in annotated_classes:
+            base_cls = annotated_classes[super_]
+            break
+
+    annotated_classes[cls] = anno_cls = cast(
+        Type[Annotated],
+        type("Annotated%s" % cls.__name__, (base_cls, cls), {}),
+    )
+    globals()["Annotated%s" % cls.__name__] = anno_cls
+
+    if "_traverse_internals" in cls.__dict__:
+        anno_cls._traverse_internals = list(cls._traverse_internals) + [
+            ("_annotations", InternalTraversal.dp_annotations_key)
+        ]
+    elif cls.__dict__.get("inherit_cache", False):
+        anno_cls._traverse_internals = list(cls._traverse_internals) + [
+            ("_annotations", InternalTraversal.dp_annotations_key)
+        ]
+
+    # some classes include this even if they have traverse_internals
+    # e.g. BindParameter, add it if present.
+    if cls.__dict__.get("inherit_cache", False):
+        anno_cls.inherit_cache = True  # type: ignore
+    elif "inherit_cache" in cls.__dict__:
+        anno_cls.inherit_cache = cls.__dict__["inherit_cache"]  # type: ignore
+
+    anno_cls._is_column_operators = issubclass(cls, operators.ColumnOperators)
+
+    return anno_cls
+
+
+def _prepare_annotations(
+    target_hierarchy: Type[SupportsWrappingAnnotations],
+    base_cls: Type[Annotated],
+) -> None:
+    for cls in util.walk_subclasses(target_hierarchy):
+        _new_annotation_type(cls, base_cls)