path: root/test/orm/declarative/test_tm_future_annotations_sync.py

"""This file is automatically generated from the file
'test/orm/declarative/test_typed_mapping.py'
by the 'tools/sync_test_files.py' script.

Do not edit manually, any change will be lost.
"""  # noqa: E501

from __future__ import annotations

import collections.abc
import dataclasses
import datetime
from decimal import Decimal
import enum
import typing
from typing import Any
from typing import ClassVar
from typing import Dict
from typing import Generic
from typing import List
from typing import NewType
from typing import Optional
from typing import Set
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
import uuid

from typing_extensions import get_args as get_args
from typing_extensions import Literal as Literal

from sqlalchemy import BIGINT
from sqlalchemy import BigInteger
from sqlalchemy import Column
from sqlalchemy import DateTime
from sqlalchemy import exc as sa_exc
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Identity
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import JSON
from sqlalchemy import Numeric
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy import types
from sqlalchemy import VARCHAR
from sqlalchemy.exc import ArgumentError
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import as_declarative
from sqlalchemy.orm import composite
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import deferred
from sqlalchemy.orm import DynamicMapped
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import undefer
from sqlalchemy.orm import WriteOnlyMapped
from sqlalchemy.orm.collections import attribute_keyed_dict
from sqlalchemy.orm.collections import KeyFuncDict
from sqlalchemy.schema import CreateTable
from sqlalchemy.sql.sqltypes import Enum
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_not
from sqlalchemy.testing import is_true
from sqlalchemy.testing import Variation
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.util import compat
from sqlalchemy.util.typing import Annotated


def expect_annotation_syntax_error(name):
    return expect_raises_message(
        sa_exc.ArgumentError,
        f'Type annotation for "{name}" '
        "can't be correctly interpreted for "
        "Annotated Declarative Table form.  ",
    )


class DeclarativeBaseTest(fixtures.TestBase):
    def test_class_getitem_as_declarative(self):
        T = TypeVar("T", bound="CommonBase")  # noqa

        class CommonBase(Generic[T]):
            @classmethod
            def boring(cls: Type[T]) -> Type[T]:
                return cls

            @classmethod
            def more_boring(cls: Type[T]) -> int:
                return 27

        @as_declarative()
        class Base(CommonBase[T]):
            foo = 1

        class Tab(Base["Tab"]):
            __tablename__ = "foo"

            # old mypy plugin use
            a: int = Column(Integer, primary_key=True)

        eq_(Tab.foo, 1)
        is_(Tab.__table__, inspect(Tab).local_table)
        eq_(Tab.boring(), Tab)
        eq_(Tab.more_boring(), 27)

        with expect_raises(AttributeError):
            Tab.non_existent


class MappedColumnTest(fixtures.TestBase, testing.AssertsCompiledSQL):
    __dialect__ = "default"

    @testing.combinations(
        "default", "insert_default", argnames="use_paramname"
    )
    @testing.combinations(True, False, argnames="use_none")
    def test_col_defaults(self, use_paramname, use_none, decl_base):
        class Foo(decl_base):
            __tablename__ = "foo"

            id: Mapped[int] = mapped_column(primary_key=True)

            data: Mapped[int] = mapped_column(
                **{use_paramname: None if use_none else 5}
            )

        if use_none:
            assert not Foo.__table__.c.data.default
        else:
            eq_(Foo.__table__.c.data.default.arg, 5)

    def test_legacy_declarative_base(self):
        typ = VARCHAR(50)
        Base = declarative_base(type_annotation_map={str: typ})

        class MyClass(Base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str]
            x: Mapped[int]

        is_(MyClass.__table__.c.data.type, typ)
        is_true(MyClass.__table__.c.id.primary_key)

    @testing.combinations(
        (BIGINT(),),
        (BIGINT,),
        (Integer().with_variant(BIGINT, "default")),
        (Integer().with_variant(BIGINT(), "default")),
        (BIGINT().with_variant(String(), "some_other_dialect")),
    )
    def test_type_map_varieties(self, typ):

        Base = declarative_base(type_annotation_map={int: typ})

        class MyClass(Base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            x: Mapped[int]
            y: Mapped[int] = mapped_column()
            z: Mapped[int] = mapped_column(typ)

        self.assert_compile(
            CreateTable(MyClass.__table__),
            "CREATE TABLE mytable (id BIGINT NOT NULL, "
            "x BIGINT NOT NULL, y BIGINT NOT NULL, z BIGINT NOT NULL, "
            "PRIMARY KEY (id))",
        )

    def test_required_no_arg(self, decl_base):
        with expect_raises_message(
            sa_exc.ArgumentError,
            r"Python typing annotation is required for attribute "
            r'"A.data" when primary '
            r'argument\(s\) for "MappedColumn" construct are None or '
            r"not present",
        ):

            class A(decl_base):
                __tablename__ = "a"

                id: Mapped[int] = mapped_column(primary_key=True)
                data = mapped_column()

    @testing.variation("case", ["key", "name", "both"])
    @testing.variation("deferred", [True, False])
    @testing.variation("use_add_property", [True, False])
    def test_separate_name(self, decl_base, case, deferred, use_add_property):
        if case.key:
            args = {"key": "data_"}
        elif case.name:
            args = {"name": "data_"}
        else:
            args = {"name": "data_", "key": "data_"}

        if deferred:
            args["deferred"] = True

        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)

            if not use_add_property:
                data: Mapped[str] = mapped_column(**args)

        if use_add_property:
            args["type_"] = String()
            A.data = mapped_column(**args)

        assert not hasattr(A, "data_")
        is_(A.data.property.expression, A.__table__.c.data_)
        eq_(A.__table__.c.data_.key, "data_")

    def test_construct_rhs(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id = mapped_column("id", Integer, primary_key=True)
            name = mapped_column(String(50))

        self.assert_compile(
            select(User), "SELECT users.id, users.name FROM users"
        )
        eq_(User.__mapper__.primary_key, (User.__table__.c.id,))

    def test_construct_lhs(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()
            data: Mapped[Optional[str]] = mapped_column()

        self.assert_compile(
            select(User), "SELECT users.id, users.name, users.data FROM users"
        )
        eq_(User.__mapper__.primary_key, (User.__table__.c.id,))
        is_false(User.__table__.c.id.nullable)
        is_false(User.__table__.c.name.nullable)
        is_true(User.__table__.c.data.nullable)

    def test_construct_lhs_omit_mapped_column(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str]
            data: Mapped[Optional[str]]
            x: Mapped[int]
            y: Mapped[int]
            created_at: Mapped[datetime.datetime]

        self.assert_compile(
            select(User),
            "SELECT users.id, users.name, users.data, users.x, "
            "users.y, users.created_at FROM users",
        )
        eq_(User.__mapper__.primary_key, (User.__table__.c.id,))
        is_false(User.__table__.c.id.nullable)
        is_false(User.__table__.c.name.nullable)
        is_true(User.__table__.c.data.nullable)
        assert isinstance(User.__table__.c.created_at.type, DateTime)

    def test_i_have_a_classvar_on_my_class(self, decl_base):
        class MyClass(decl_base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column(default="some default")

            status: ClassVar[int]

        m1 = MyClass(id=1, data=5)
        assert "status" not in inspect(m1).mapper.attrs

    def test_i_have_plain_or_column_attrs_on_my_class_w_values(
        self, decl_base
    ):
        class MyClass(decl_base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column(default="some default")

            old_column: str = Column(String)

            # we assume this is intentional
            status: int = 5

        # it's mapped too
        assert "old_column" in inspect(MyClass).attrs

    def test_i_have_plain_attrs_on_my_class_disallowed(self, decl_base):
        with expect_annotation_syntax_error("MyClass.status"):

            class MyClass(decl_base):
                __tablename__ = "mytable"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: Mapped[str] = mapped_column(default="some default")

                # we assume this is not intentional.  because I made the
                # same mistake myself :)
                status: int

    def test_i_have_plain_attrs_on_my_class_allowed(self, decl_base):
        class MyClass(decl_base):
            __tablename__ = "mytable"
            __allow_unmapped__ = True

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column(default="some default")

            status: int

    def test_allow_unmapped_on_mixin(self, decl_base):
        class AllowsUnmapped:
            __allow_unmapped__ = True

        class MyClass(AllowsUnmapped, decl_base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column(default="some default")

            status: int

    def test_allow_unmapped_on_base(self):
        class Base(DeclarativeBase):
            __allow_unmapped__ = True

        class MyClass(Base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column(default="some default")

            status: int

    def test_column_default(self, decl_base):
        class MyClass(decl_base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column(default="some default")

        mc = MyClass()
        assert "data" not in mc.__dict__

        eq_(MyClass.__table__.c.data.default.arg, "some default")

    def test_anno_w_fixed_table(self, decl_base):
        users = Table(
            "users",
            decl_base.metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(50), nullable=False),
            Column("data", String(50)),
            Column("x", Integer),
            Column("y", Integer),
            Column("created_at", DateTime),
        )

        class User(decl_base):
            __table__ = users

            id: Mapped[int]
            name: Mapped[str]
            data: Mapped[Optional[str]]
            x: Mapped[int]
            y: Mapped[int]
            created_at: Mapped[datetime.datetime]

        self.assert_compile(
            select(User),
            "SELECT users.id, users.name, users.data, users.x, "
            "users.y, users.created_at FROM users",
        )
        eq_(User.__mapper__.primary_key, (User.__table__.c.id,))
        is_false(User.__table__.c.id.nullable)
        is_false(User.__table__.c.name.nullable)
        is_true(User.__table__.c.data.nullable)
        assert isinstance(User.__table__.c.created_at.type, DateTime)

    def test_construct_lhs_type_missing(self, decl_base):
        global MyClass

        class MyClass:
            pass

        with expect_raises_message(
            sa_exc.ArgumentError,
            "Could not locate SQLAlchemy Core type for Python type "
            ".*MyClass.* inside the 'data' attribute Mapped annotation",
        ):

            class User(decl_base):
                __tablename__ = "users"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: Mapped[MyClass] = mapped_column()

    def test_construct_lhs_sqlalchemy_type(self, decl_base):

        with expect_raises_message(
            sa_exc.ArgumentError,
            "The type provided inside the 'data' attribute Mapped "
            "annotation is the SQLAlchemy type .*BigInteger.*. Expected "
            "a Python type instead",
        ):

            class User(decl_base):
                __tablename__ = "users"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: Mapped[BigInteger] = mapped_column()

    def test_construct_rhs_type_override_lhs(self, decl_base):
        class Element(decl_base):
            __tablename__ = "element"

            id: Mapped[int] = mapped_column(BIGINT, primary_key=True)

        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(BIGINT, primary_key=True)
            other_id: Mapped[int] = mapped_column(ForeignKey("element.id"))
            data: Mapped[int] = mapped_column()

        # exact class test
        is_(User.__table__.c.id.type.__class__, BIGINT)
        is_(User.__table__.c.other_id.type.__class__, BIGINT)
        is_(User.__table__.c.data.type.__class__, Integer)

    @testing.combinations(True, False, argnames="include_rhs_type")
    @testing.combinations(True, False, argnames="use_mixin")
    def test_construct_nullability_overrides(
        self, decl_base, include_rhs_type, use_mixin
    ):

        if include_rhs_type:
            args = (String,)
        else:
            args = ()

        global anno_str, anno_str_optional, anno_str_mc
        global anno_str_optional_mc, anno_str_mc_nullable
        global anno_str_optional_mc_notnull
        global newtype_str

        anno_str = Annotated[str, 50]
        anno_str_optional = Annotated[Optional[str], 30]

        newtype_str = NewType("MyType", str)

        anno_str_mc = Annotated[str, mapped_column()]
        anno_str_optional_mc = Annotated[Optional[str], mapped_column()]
        anno_str_mc_nullable = Annotated[str, mapped_column(nullable=True)]
        anno_str_optional_mc_notnull = Annotated[
            Optional[str], mapped_column(nullable=False)
        ]

        decl_base.registry.update_type_annotation_map(
            {
                anno_str: String(50),
                anno_str_optional: String(30),
                newtype_str: String(40),
            }
        )

        if TYPE_CHECKING:

            class user_base:
                pass

        else:
            if use_mixin:
                user_base = object
            else:
                user_base = decl_base

        class UserPossibleMixin(user_base):
            if not use_mixin:
                __tablename__ = "users"

                id: Mapped[int] = mapped_column(primary_key=True)  # noqa: A001

            lnnl_rndf: Mapped[str] = mapped_column(*args)
            lnnl_rnnl: Mapped[str] = mapped_column(*args, nullable=False)
            lnnl_rnl: Mapped[str] = mapped_column(*args, nullable=True)
            lnl_rndf: Mapped[Optional[str]] = mapped_column(*args)
            lnl_rnnl: Mapped[Optional[str]] = mapped_column(
                *args, nullable=False
            )
            lnl_rnl: Mapped[Optional[str]] = mapped_column(
                *args, nullable=True
            )

            # test #9177 cases
            anno_1a: Mapped[anno_str] = mapped_column(*args)
            anno_1b: Mapped[anno_str] = mapped_column(*args, nullable=True)
            anno_1c: Mapped[anno_str] = mapped_column(*args, deferred=True)
            anno_1d: Mapped[anno_str] = mapped_column(
                *args, deferred=True, deferred_group="mygroup"
            )

            anno_2a: Mapped[anno_str_optional] = mapped_column(*args)
            anno_2b: Mapped[anno_str_optional] = mapped_column(
                *args, nullable=False
            )

            anno_3a: Mapped[anno_str_mc] = mapped_column(*args)
            anno_3b: Mapped[anno_str_mc] = mapped_column(*args, nullable=True)
            anno_3c: Mapped[Optional[anno_str_mc]] = mapped_column(*args)

            anno_4a: Mapped[anno_str_optional_mc] = mapped_column(*args)
            anno_4b: Mapped[anno_str_optional_mc] = mapped_column(
                *args, nullable=False
            )

            anno_5a: Mapped[anno_str_mc_nullable] = mapped_column(*args)
            anno_5b: Mapped[anno_str_mc_nullable] = mapped_column(
                *args, nullable=False
            )

            anno_6a: Mapped[anno_str_optional_mc_notnull] = mapped_column(
                *args
            )
            anno_6b: Mapped[anno_str_optional_mc_notnull] = mapped_column(
                *args, nullable=True
            )

            newtype_1a: Mapped[newtype_str] = mapped_column(*args)
            newtype_1b: Mapped[newtype_str] = mapped_column(
                *args, nullable=True
            )

        if use_mixin:

            class User(UserPossibleMixin, decl_base):
                __tablename__ = "users"

                id: Mapped[int] = mapped_column(primary_key=True)

        else:
            User = UserPossibleMixin

        eq_(User.anno_1b.property.deferred, False)
        eq_(User.anno_1c.property.deferred, True)
        eq_(User.anno_1d.property.group, "mygroup")

        is_false(User.__table__.c.lnnl_rndf.nullable)
        is_false(User.__table__.c.lnnl_rnnl.nullable)
        is_true(User.__table__.c.lnnl_rnl.nullable)

        is_true(User.__table__.c.lnl_rndf.nullable)
        is_false(User.__table__.c.lnl_rnnl.nullable)
        is_true(User.__table__.c.lnl_rnl.nullable)

        is_false(User.__table__.c.anno_1a.nullable)
        is_true(User.__table__.c.anno_1b.nullable)
        is_true(User.__table__.c.anno_2a.nullable)
        is_false(User.__table__.c.anno_2b.nullable)
        is_false(User.__table__.c.anno_3a.nullable)
        is_true(User.__table__.c.anno_3b.nullable)
        is_true(User.__table__.c.anno_3c.nullable)
        is_true(User.__table__.c.anno_4a.nullable)
        is_false(User.__table__.c.anno_4b.nullable)
        is_true(User.__table__.c.anno_5a.nullable)
        is_false(User.__table__.c.anno_5b.nullable)
        is_false(User.__table__.c.anno_6a.nullable)
        is_true(User.__table__.c.anno_6b.nullable)

        # test #8410
        is_false(User.__table__.c.lnnl_rndf._copy().nullable)
        is_false(User.__table__.c.lnnl_rnnl._copy().nullable)
        is_true(User.__table__.c.lnnl_rnl._copy().nullable)
        is_true(User.__table__.c.lnl_rndf._copy().nullable)
        is_false(User.__table__.c.lnl_rnnl._copy().nullable)
        is_true(User.__table__.c.lnl_rnl._copy().nullable)

    def test_fwd_refs(self, decl_base: Type[DeclarativeBase]):
        class MyClass(decl_base):
            __tablename__ = "my_table"

            id: Mapped["int"] = mapped_column(primary_key=True)
            data_one: Mapped["str"]

    def test_pep593_types_as_typemap_keys(
        self, decl_base: Type[DeclarativeBase]
    ):
        """neat!!!"""
        global str50, str30, opt_str50, opt_str30

        str50 = Annotated[str, 50]
        str30 = Annotated[str, 30]
        opt_str50 = Optional[str50]
        opt_str30 = Optional[str30]

        decl_base.registry.update_type_annotation_map(
            {str50: String(50), str30: String(30)}
        )

        class MyClass(decl_base):
            __tablename__ = "my_table"

            id: Mapped[str50] = mapped_column(primary_key=True)
            data_one: Mapped[str30]
            data_two: Mapped[opt_str30]
            data_three: Mapped[str50]
            data_four: Mapped[opt_str50]
            data_five: Mapped[str]
            data_six: Mapped[Optional[str]]

        eq_(MyClass.__table__.c.data_one.type.length, 30)
        is_false(MyClass.__table__.c.data_one.nullable)
        eq_(MyClass.__table__.c.data_two.type.length, 30)
        is_true(MyClass.__table__.c.data_two.nullable)
        eq_(MyClass.__table__.c.data_three.type.length, 50)

    def test_pep484_newtypes_as_typemap_keys(
        self, decl_base: Type[DeclarativeBase]
    ):

        global str50, str30, str3050

        str50 = NewType("str50", str)
        str30 = NewType("str30", str)
        str3050 = NewType("str30", str50)

        decl_base.registry.update_type_annotation_map(
            {str50: String(50), str30: String(30), str3050: String(150)}
        )

        class MyClass(decl_base):
            __tablename__ = "my_table"

            id: Mapped[str50] = mapped_column(primary_key=True)
            data_one: Mapped[str30]
            data_two: Mapped[str50]
            data_three: Mapped[Optional[str30]]
            data_four: Mapped[str3050]

        eq_(MyClass.__table__.c.data_one.type.length, 30)
        is_false(MyClass.__table__.c.data_one.nullable)

        eq_(MyClass.__table__.c.data_two.type.length, 50)
        is_false(MyClass.__table__.c.data_two.nullable)

        eq_(MyClass.__table__.c.data_three.type.length, 30)
        is_true(MyClass.__table__.c.data_three.nullable)

        eq_(MyClass.__table__.c.data_four.type.length, 150)
        is_false(MyClass.__table__.c.data_four.nullable)

    def test_extract_base_type_from_pep593(
        self, decl_base: Type[DeclarativeBase]
    ):
        """base type is extracted from an Annotated structure if not otherwise
        in the type lookup dictionary"""

        class MyClass(decl_base):
            __tablename__ = "my_table"

            id: Mapped[Annotated[Annotated[int, "q"], "t"]] = mapped_column(
                primary_key=True
            )

        is_(MyClass.__table__.c.id.type._type_affinity, Integer)

    def test_extract_sqla_from_pep593_not_yet(
        self, decl_base: Type[DeclarativeBase]
    ):
        """https://twitter.com/zzzeek/status/1536693554621341697"""

        global SomeRelated

        class SomeRelated(decl_base):
            __tablename__: ClassVar[Optional[str]] = "some_related"
            id: Mapped["int"] = mapped_column(primary_key=True)

        with expect_raises_message(
            NotImplementedError,
            r"Use of the \<class 'sqlalchemy.orm."
            r"relationships.Relationship'\> construct inside of an Annotated "
            r"object is not yet supported.",
        ):

            class MyClass(decl_base):
                __tablename__ = "my_table"

                id: Mapped["int"] = mapped_column(primary_key=True)
                data_one: Mapped[Annotated["SomeRelated", relationship()]]

    def test_extract_sqla_from_pep593_plain(
        self, decl_base: Type[DeclarativeBase]
    ):
        """extraction of mapped_column() from the Annotated type

        https://twitter.com/zzzeek/status/1536693554621341697"""
        global intpk, strnone, str30nullable
        global opt_strnone, opt_str30

        intpk = Annotated[int, mapped_column(primary_key=True)]

        strnone = Annotated[str, mapped_column()]  # str -> NOT NULL
        str30nullable = Annotated[
            str, mapped_column(String(30), nullable=True)  # nullable -> NULL
        ]
        opt_strnone = Optional[strnone]  # Optional[str] -> NULL
        opt_str30 = Optional[str30nullable]  # nullable -> NULL

        class MyClass(decl_base):
            __tablename__ = "my_table"

            id: Mapped[intpk]

            data_one: Mapped[strnone]
            data_two: Mapped[str30nullable]
            data_three: Mapped[opt_strnone]
            data_four: Mapped[opt_str30]

        class MyOtherClass(decl_base):
            __tablename__ = "my_other_table"

            id: Mapped[intpk]

            data_one: Mapped[strnone]
            data_two: Mapped[str30nullable]
            data_three: Mapped[opt_strnone]
            data_four: Mapped[opt_str30]

        for cls in MyClass, MyOtherClass:
            table = cls.__table__
            assert table is not None

            is_(table.c.id.primary_key, True)
            is_(table.c.id.table, table)

            eq_(table.c.data_one.type.length, None)
            eq_(table.c.data_two.type.length, 30)
            eq_(table.c.data_three.type.length, None)

            is_false(table.c.data_one.nullable)
            is_true(table.c.data_two.nullable)
            is_true(table.c.data_three.nullable)
            is_true(table.c.data_four.nullable)

    def test_extract_sqla_from_pep593_mixin(
        self, decl_base: Type[DeclarativeBase]
    ):
        """extraction of mapped_column() from the Annotated type

        https://twitter.com/zzzeek/status/1536693554621341697"""

        global intpk, strnone, str30nullable
        global opt_strnone, opt_str30
        intpk = Annotated[int, mapped_column(primary_key=True)]

        strnone = Annotated[str, mapped_column()]  # str -> NOT NULL
        str30nullable = Annotated[
            str, mapped_column(String(30), nullable=True)  # nullable -> NULL
        ]
        opt_strnone = Optional[strnone]  # Optional[str] -> NULL
        opt_str30 = Optional[str30nullable]  # nullable -> NULL

        class HasPk:
            id: Mapped[intpk]

            data_one: Mapped[strnone]
            data_two: Mapped[str30nullable]

        class MyClass(HasPk, decl_base):
            __tablename__ = "my_table"

            data_three: Mapped[opt_strnone]
            data_four: Mapped[opt_str30]

        table = MyClass.__table__
        assert table is not None

        is_(table.c.id.primary_key, True)
        is_(table.c.id.table, table)

        eq_(table.c.data_one.type.length, None)
        eq_(table.c.data_two.type.length, 30)
        eq_(table.c.data_three.type.length, None)

        is_false(table.c.data_one.nullable)
        is_true(table.c.data_two.nullable)
        is_true(table.c.data_three.nullable)
        is_true(table.c.data_four.nullable)

    @testing.variation("to_assert", ["ddl", "fkcount", "references"])
    @testing.variation("assign_blank", [True, False])
    def test_extract_fk_col_from_pep593(
        self, decl_base: Type[DeclarativeBase], to_assert, assign_blank
    ):
        global intpk, element_ref
        intpk = Annotated[int, mapped_column(primary_key=True)]
        element_ref = Annotated[int, mapped_column(ForeignKey("element.id"))]

        class Element(decl_base):
            __tablename__ = "element"

            id: Mapped[intpk]

        class RefElementOne(decl_base):
            __tablename__ = "refone"

            id: Mapped[intpk]

            if assign_blank:
                other_id: Mapped[element_ref] = mapped_column()
            else:
                other_id: Mapped[element_ref]

        class RefElementTwo(decl_base):
            __tablename__ = "reftwo"

            id: Mapped[intpk]
            if assign_blank:
                some_id: Mapped[element_ref] = mapped_column()
            else:
                some_id: Mapped[element_ref]

        assert Element.__table__ is not None
        assert RefElementOne.__table__ is not None
        assert RefElementTwo.__table__ is not None

        if to_assert.fkcount:
            # test #9766
            eq_(len(RefElementOne.__table__.c.other_id.foreign_keys), 1)
            eq_(len(RefElementTwo.__table__.c.some_id.foreign_keys), 1)
        elif to_assert.references:
            is_true(
                RefElementOne.__table__.c.other_id.references(
                    Element.__table__.c.id
                )
            )
            is_true(
                RefElementTwo.__table__.c.some_id.references(
                    Element.__table__.c.id
                )
            )

        elif to_assert.ddl:
            self.assert_compile(
                CreateTable(RefElementOne.__table__),
                "CREATE TABLE refone "
                "(id INTEGER NOT NULL, other_id INTEGER NOT NULL, "
                "PRIMARY KEY (id), "
                "FOREIGN KEY(other_id) REFERENCES element (id))",
            )
            self.assert_compile(
                CreateTable(RefElementTwo.__table__),
                "CREATE TABLE reftwo "
                "(id INTEGER NOT NULL, some_id INTEGER NOT NULL, "
                "PRIMARY KEY (id), "
                "FOREIGN KEY(some_id) REFERENCES element (id))",
            )
        else:
            to_assert.fail()

    @testing.combinations(
        (collections.abc.Sequence, (str,), testing.requires.python310),
        (collections.abc.MutableSequence, (str,), testing.requires.python310),
        (collections.abc.Mapping, (str, str), testing.requires.python310),
        (
            collections.abc.MutableMapping,
            (str, str),
            testing.requires.python310,
        ),
        (typing.Mapping, (str, str), testing.requires.python310),
        (typing.MutableMapping, (str, str), testing.requires.python310),
        (typing.Sequence, (str,)),
        (typing.MutableSequence, (str,)),
        (list, (str,), testing.requires.python310),
        (
            List,
            (str,),
        ),
        (dict, (str, str), testing.requires.python310),
        (
            Dict,
            (str, str),
        ),
        id_="sa",
    )
    def test_extract_generic_from_pep593(self, container_typ, args):
        """test #9099"""

        global TestType
        TestType = Annotated[container_typ[args], 0]

        class Base(DeclarativeBase):
            type_annotation_map = {TestType: JSON()}

        class MyClass(Base):
            __tablename__ = "my_table"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[TestType] = mapped_column()

        is_(MyClass.__table__.c.data.type._type_affinity, JSON)

    @testing.combinations(
        ("default", lambda ctx: 10),
        ("default", func.foo()),
        ("onupdate", lambda ctx: 10),
        ("onupdate", func.foo()),
        ("server_onupdate", func.foo()),
        ("server_default", func.foo()),
        ("server_default", Identity()),
        ("nullable", True),
        ("nullable", False),
        ("type", BigInteger()),
        ("index", True),
        ("unique", True),
        argnames="paramname, value",
    )
    @testing.combinations(True, False, argnames="optional")
    @testing.combinations(True, False, argnames="include_existing_col")
    def test_combine_args_from_pep593(
        self,
        decl_base: Type[DeclarativeBase],
        paramname,
        value,
        include_existing_col,
        optional,
    ):
        global intpk, element_ref
        intpk = Annotated[int, mapped_column(primary_key=True)]

        args = []
        params = {}
        if paramname == "type":
            args.append(value)
        else:
            params[paramname] = value

        element_ref = Annotated[int, mapped_column(*args, **params)]
        if optional:
            element_ref = Optional[element_ref]

        class Element(decl_base):
            __tablename__ = "element"

            id: Mapped[intpk]

            if include_existing_col:
                data: Mapped[element_ref] = mapped_column()
            else:
                data: Mapped[element_ref]

        data_col = Element.__table__.c.data
        if paramname in (
            "default",
            "onupdate",
            "server_default",
            "server_onupdate",
        ):
            default = getattr(data_col, paramname)
            if default.is_server_default and default.has_argument:
                is_(default.arg, value)
            is_(default.column, data_col)
        elif paramname == "type":
            assert type(data_col.type) is type(value)
        else:
            is_(getattr(data_col, paramname), value)

            # test _copy() for #8410
            is_(getattr(data_col._copy(), paramname), value)

        sd = data_col.server_default
        if sd is not None and isinstance(sd, Identity):
            if paramname == "nullable" and value:
                is_(data_col.nullable, True)
            else:
                is_(data_col.nullable, False)
        elif paramname != "nullable":
            is_(data_col.nullable, optional)
        else:
            is_(data_col.nullable, value)

    @testing.combinations(True, False, argnames="specify_identity")
    @testing.combinations(True, False, None, argnames="specify_nullable")
    @testing.combinations(True, False, argnames="optional")
    @testing.combinations(True, False, argnames="include_existing_col")
    def test_combine_args_from_pep593_identity_nullable(
        self,
        decl_base: Type[DeclarativeBase],
        specify_identity,
        specify_nullable,
        optional,
        include_existing_col,
    ):
        global intpk, element_ref
        intpk = Annotated[int, mapped_column(primary_key=True)]

        if specify_identity:
            args = [Identity()]
        else:
            args = []

        if specify_nullable is not None:
            params = {"nullable": specify_nullable}
        else:
            params = {}

        element_ref = Annotated[int, mapped_column(*args, **params)]
        if optional:
            element_ref = Optional[element_ref]

        class Element(decl_base):
            __tablename__ = "element"

            id: Mapped[intpk]

            if include_existing_col:
                data: Mapped[element_ref] = mapped_column()
            else:
                data: Mapped[element_ref]

        # test identity + _copy() for #8410
        for col in (
            Element.__table__.c.data,
            Element.__table__.c.data._copy(),
        ):
            if specify_nullable is True:
                is_(col.nullable, True)
            elif specify_identity:
                is_(col.nullable, False)
            elif specify_nullable is False:
                is_(col.nullable, False)
            elif not optional:
                is_(col.nullable, False)
            else:
                is_(col.nullable, True)

    @testing.combinations(
        ("default", lambda ctx: 10, lambda ctx: 15),
        ("default", func.foo(), func.bar()),
        ("onupdate", lambda ctx: 10, lambda ctx: 15),
        ("onupdate", func.foo(), func.bar()),
        ("server_onupdate", func.foo(), func.bar()),
        ("server_default", func.foo(), func.bar()),
        ("nullable", True, False),
        ("nullable", False, True),
        ("type", BigInteger(), Numeric()),
        argnames="paramname, value, override_value",
    )
    def test_dont_combine_args_from_pep593(
        self,
        decl_base: Type[DeclarativeBase],
        paramname,
        value,
        override_value,
    ):
        global intpk, element_ref
        intpk = Annotated[int, mapped_column(primary_key=True)]

        args = []
        params = {}
        override_args = []
        override_params = {}
        if paramname == "type":
            args.append(value)
            override_args.append(override_value)
        else:
            params[paramname] = value
            if paramname == "default":
                override_params["insert_default"] = override_value
            else:
                override_params[paramname] = override_value

        element_ref = Annotated[int, mapped_column(*args, **params)]

        class Element(decl_base):
            __tablename__ = "element"

            id: Mapped[intpk]

            data: Mapped[element_ref] = mapped_column(
                *override_args, **override_params
            )

        if paramname in (
            "default",
            "onupdate",
            "server_default",
            "server_onupdate",
        ):
            default = getattr(Element.__table__.c.data, paramname)
            is_(default.arg, override_value)
            is_(default.column, Element.__table__.c.data)
        elif paramname == "type":
            assert type(Element.__table__.c.data.type) is type(override_value)
        else:
            is_(getattr(Element.__table__.c.data, paramname), override_value)

    def test_unions(self):
        our_type = Numeric(10, 2)

        class Base(DeclarativeBase):
            type_annotation_map = {Union[float, Decimal]: our_type}

        class User(Base):
            __tablename__ = "users"
            __table__: Table

            id: Mapped[int] = mapped_column(primary_key=True)

            data: Mapped[Union[float, Decimal]] = mapped_column()
            reverse_data: Mapped[Union[Decimal, float]] = mapped_column()

            optional_data: Mapped[
                Optional[Union[float, Decimal]]
            ] = mapped_column()

            # use Optional directly
            reverse_optional_data: Mapped[
                Optional[Union[Decimal, float]]
            ] = mapped_column()

            # use Union with None, same as Optional but presents differently
            # (Optional object with __origin__ Union vs. Union)
            reverse_u_optional_data: Mapped[
                Union[Decimal, float, None]
            ] = mapped_column()

            float_data: Mapped[float] = mapped_column()
            decimal_data: Mapped[Decimal] = mapped_column()

            if compat.py310:
                pep604_data: Mapped[float | Decimal] = mapped_column()
                pep604_reverse: Mapped[Decimal | float] = mapped_column()
                pep604_optional: Mapped[
                    Decimal | float | None
                ] = mapped_column()
                pep604_data_fwd: Mapped["float | Decimal"] = mapped_column()
                pep604_reverse_fwd: Mapped["Decimal | float"] = mapped_column()
                pep604_optional_fwd: Mapped[
                    "Decimal | float | None"
                ] = mapped_column()

        is_(User.__table__.c.data.type, our_type)
        is_false(User.__table__.c.data.nullable)
        is_(User.__table__.c.reverse_data.type, our_type)
        is_(User.__table__.c.optional_data.type, our_type)
        is_true(User.__table__.c.optional_data.nullable)

        is_(User.__table__.c.reverse_optional_data.type, our_type)
        is_(User.__table__.c.reverse_u_optional_data.type, our_type)
        is_true(User.__table__.c.reverse_optional_data.nullable)
        is_true(User.__table__.c.reverse_u_optional_data.nullable)

        is_(User.__table__.c.float_data.type, our_type)
        is_(User.__table__.c.decimal_data.type, our_type)

        if compat.py310:
            for suffix in ("", "_fwd"):
                data_col = User.__table__.c[f"pep604_data{suffix}"]
                reverse_col = User.__table__.c[f"pep604_reverse{suffix}"]
                optional_col = User.__table__.c[f"pep604_optional{suffix}"]
                is_(data_col.type, our_type)
                is_false(data_col.nullable)
                is_(reverse_col.type, our_type)
                is_false(reverse_col.nullable)
                is_(optional_col.type, our_type)
                is_true(optional_col.nullable)

    @testing.combinations(
        ("not_optional",),
        ("optional",),
        ("optional_fwd_ref",),
        ("union_none",),
        ("pep604", testing.requires.python310),
        ("pep604_fwd_ref", testing.requires.python310),
        argnames="optional_on_json",
    )
    @testing.combinations(
        "include_mc_type", "derive_from_anno", argnames="include_mc_type"
    )
    def test_optional_styles_nested_brackets(
        self, optional_on_json, include_mc_type
    ):
        class Base(DeclarativeBase):
            if testing.requires.python310.enabled:
                type_annotation_map = {
                    Dict[str, str]: JSON,
                    dict[str, str]: JSON,
                }
            else:
                type_annotation_map = {
                    Dict[str, str]: JSON,
                }

        if include_mc_type == "include_mc_type":
            mc = mapped_column(JSON)
        else:
            mc = mapped_column()

        class A(Base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column()

            if optional_on_json == "not_optional":
                json: Mapped[Dict[str, str]] = mapped_column()  # type: ignore
            elif optional_on_json == "optional":
                json: Mapped[Optional[Dict[str, str]]] = mc
            elif optional_on_json == "optional_fwd_ref":
                json: Mapped["Optional[Dict[str, str]]"] = mc
            elif optional_on_json == "union_none":
                json: Mapped[Union[Dict[str, str], None]] = mc
            elif optional_on_json == "pep604":
                json: Mapped[dict[str, str] | None] = mc
            elif optional_on_json == "pep604_fwd_ref":
                json: Mapped["dict[str, str] | None"] = mc

        is_(A.__table__.c.json.type._type_affinity, JSON)
        if optional_on_json == "not_optional":
            is_false(A.__table__.c.json.nullable)
        else:
            is_true(A.__table__.c.json.nullable)

    @testing.variation("optional", [True, False])
    @testing.variation("provide_type", [True, False])
    @testing.variation("add_to_type_map", [True, False])
    def test_recursive_type(
        self, decl_base, optional, provide_type, add_to_type_map
    ):
        """test #9553"""

        global T

        T = Dict[str, Optional["T"]]

        if not provide_type and not add_to_type_map:
            with expect_raises_message(
                sa_exc.ArgumentError,
                r"Could not locate SQLAlchemy.*" r".*ForwardRef\('T'\).*",
            ):

                class TypeTest(decl_base):
                    __tablename__ = "my_table"

                    id: Mapped[int] = mapped_column(primary_key=True)
                    if optional:
                        type_test: Mapped[Optional[T]] = mapped_column()
                    else:
                        type_test: Mapped[T] = mapped_column()

            return

        else:
            if add_to_type_map:
                decl_base.registry.update_type_annotation_map({T: JSON()})

            class TypeTest(decl_base):
                __tablename__ = "my_table"

                id: Mapped[int] = mapped_column(primary_key=True)

                if add_to_type_map:
                    if optional:
                        type_test: Mapped[Optional[T]] = mapped_column()
                    else:
                        type_test: Mapped[T] = mapped_column()
                else:
                    if optional:
                        type_test: Mapped[Optional[T]] = mapped_column(JSON())
                    else:
                        type_test: Mapped[T] = mapped_column(JSON())

        if optional:
            is_(TypeTest.__table__.c.type_test.nullable, True)
        else:
            is_(TypeTest.__table__.c.type_test.nullable, False)

        self.assert_compile(
            select(TypeTest),
            "SELECT my_table.id, my_table.type_test FROM my_table",
        )

    def test_missing_mapped_lhs(self, decl_base):
        with expect_annotation_syntax_error("User.name"):

            class User(decl_base):
                __tablename__ = "users"

                id: Mapped[int] = mapped_column(primary_key=True)
                name: str = mapped_column()  # type: ignore

    def test_construct_lhs_separate_name(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()
            data: Mapped[Optional[str]] = mapped_column("the_data")

        self.assert_compile(
            select(User.data), "SELECT users.the_data FROM users"
        )
        is_true(User.__table__.c.the_data.nullable)

    def test_construct_works_in_expr(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)

        class Address(decl_base):
            __tablename__ = "addresses"

            id: Mapped[int] = mapped_column(primary_key=True)
            user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

            user = relationship(User, primaryjoin=user_id == User.id)

        self.assert_compile(
            select(Address.user_id, User.id).join(Address.user),
            "SELECT addresses.user_id, users.id FROM addresses "
            "JOIN users ON addresses.user_id = users.id",
        )

    def test_construct_works_as_polymorphic_on(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)
            type: Mapped[str] = mapped_column()

            __mapper_args__ = {"polymorphic_on": type}

        decl_base.registry.configure()
        is_(User.__table__.c.type, User.__mapper__.polymorphic_on)

    def test_construct_works_as_version_id_col(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)
            version_id: Mapped[int] = mapped_column()

            __mapper_args__ = {"version_id_col": version_id}

        decl_base.registry.configure()
        is_(User.__table__.c.version_id, User.__mapper__.version_id_col)

    def test_construct_works_in_deferred(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = deferred(mapped_column())

        self.assert_compile(select(User), "SELECT users.id FROM users")
        self.assert_compile(
            select(User).options(undefer(User.data)),
            "SELECT users.id, users.data FROM users",
        )

    def test_deferred_kw(self, decl_base):
        class User(decl_base):
            __tablename__ = "users"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column(deferred=True)

        self.assert_compile(select(User), "SELECT users.id FROM users")
        self.assert_compile(
            select(User).options(undefer(User.data)),
            "SELECT users.id, users.data FROM users",
        )

    @testing.combinations(
        (str, types.String),
        (Decimal, types.Numeric),
        (float, types.Float),
        (datetime.datetime, types.DateTime),
        (uuid.UUID, types.Uuid),
        argnames="pytype_arg,sqltype",
    )
    def test_datatype_lookups(self, decl_base, pytype_arg, sqltype):
        global pytype
        pytype = pytype_arg

        class MyClass(decl_base):
            __tablename__ = "mytable"
            id: Mapped[int] = mapped_column(primary_key=True)

            data: Mapped[pytype]

        assert isinstance(MyClass.__table__.c.data.type, sqltype)

    def test_dont_ignore_unresolvable(self, decl_base):
        """test #8888"""

        with expect_raises_message(
            sa_exc.ArgumentError,
            r"Could not resolve all types within mapped annotation: "
            r"\".*Mapped\[.*fake.*\]\".  Ensure all types are written "
            r"correctly and are imported within the module in use.",
        ):

            class A(decl_base):
                __tablename__ = "a"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: Mapped["fake"]  # noqa

    def test_type_dont_mis_resolve_on_superclass(self):
        """test for #8859.

        For subclasses of a type that's in the map, don't resolve this
        by default, even though we do a search through __mro__.

        """
        global int_sub

        class int_sub(int):
            pass

        Base = declarative_base(
            type_annotation_map={
                int: Integer,
            }
        )

        with expect_raises_message(
            sa_exc.ArgumentError, "Could not locate SQLAlchemy Core type"
        ):

            class MyClass(Base):
                __tablename__ = "mytable"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: Mapped[int_sub]

    @testing.variation(
        "dict_key", ["typing", ("plain", testing.requires.python310)]
    )
    def test_type_dont_mis_resolve_on_non_generic(self, dict_key):
        """test for #8859.

        For a specific generic type with arguments, don't do any MRO
        lookup.

        """

        Base = declarative_base(
            type_annotation_map={
                dict: String,
            }
        )

        with expect_raises_message(
            sa_exc.ArgumentError, "Could not locate SQLAlchemy Core type"
        ):

            class MyClass(Base):
                __tablename__ = "mytable"

                id: Mapped[int] = mapped_column(primary_key=True)

                if dict_key.plain:
                    data: Mapped[dict[str, str]]
                elif dict_key.typing:
                    data: Mapped[Dict[str, str]]

    def test_type_secondary_resolution(self):
        class MyString(String):
            def _resolve_for_python_type(
                self, python_type, matched_type, matched_on_flattened
            ):
                return String(length=42)

        Base = declarative_base(type_annotation_map={str: MyString})

        class MyClass(Base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str]

        is_true(isinstance(MyClass.__table__.c.data.type, String))
        eq_(MyClass.__table__.c.data.type.length, 42)


class EnumOrLiteralTypeMapTest(fixtures.TestBase, testing.AssertsCompiledSQL):
    __dialect__ = "default"

    @testing.variation("use_callable", [True, False])
    @testing.variation("include_generic", [True, False])
    @testing.variation("set_native_enum", ["none", True, False])
    def test_enum_explicit(
        self, use_callable, include_generic, set_native_enum: Variation
    ):
        global FooEnum

        class FooEnum(enum.Enum):
            foo = enum.auto()
            bar = enum.auto()

        kw = {"length": 500}

        if set_native_enum.none:
            expected_native_enum = True
        elif set_native_enum.set_native_enum:
            kw["native_enum"] = True
            expected_native_enum = True
        elif set_native_enum.not_set_native_enum:
            kw["native_enum"] = False
            expected_native_enum = False
        else:
            set_native_enum.fail()

        if use_callable:
            tam = {FooEnum: Enum(FooEnum, **kw)}
        else:
            tam = {FooEnum: Enum(FooEnum, **kw)}

        if include_generic:
            tam[enum.Enum] = Enum(enum.Enum)
        Base = declarative_base(type_annotation_map=tam)

        class MyClass(Base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[FooEnum]

        is_true(isinstance(MyClass.__table__.c.data.type, Enum))
        eq_(MyClass.__table__.c.data.type.length, 500)
        is_(MyClass.__table__.c.data.type.enum_class, FooEnum)
        is_(MyClass.__table__.c.data.type.native_enum, expected_native_enum)

    @testing.variation("set_native_enum", ["none", True, False])
    def test_enum_generic(self, set_native_enum: Variation):
        """test for #8859"""
        global FooEnum

        class FooEnum(enum.Enum):
            foo = enum.auto()
            bar = enum.auto()

        kw = {"length": 42}

        if set_native_enum.none:
            expected_native_enum = True
        elif set_native_enum.set_native_enum:
            kw["native_enum"] = True
            expected_native_enum = True
        elif set_native_enum.not_set_native_enum:
            kw["native_enum"] = False
            expected_native_enum = False
        else:
            set_native_enum.fail()

        Base = declarative_base(
            type_annotation_map={enum.Enum: Enum(enum.Enum, **kw)}
        )

        class MyClass(Base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[FooEnum]

        is_true(isinstance(MyClass.__table__.c.data.type, Enum))
        eq_(MyClass.__table__.c.data.type.length, 42)
        is_(MyClass.__table__.c.data.type.enum_class, FooEnum)
        is_(MyClass.__table__.c.data.type.native_enum, expected_native_enum)

    def test_enum_default(self, decl_base):
        """test #8859.

        We now have Enum in the default SQL lookup map, in conjunction with
        a mechanism that will adapt it for a given enum type.

        This relies on a search through __mro__ for the given type,
        which in other tests we ensure does not actually function if
        we aren't dealing with Enum (or some other type that allows for
        __mro__ lookup)

        """
        global FooEnum

        class FooEnum(enum.Enum):
            foo = "foo"
            bar_value = "bar"

        class MyClass(decl_base):
            __tablename__ = "mytable"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[FooEnum]

        is_true(isinstance(MyClass.__table__.c.data.type, Enum))
        eq_(MyClass.__table__.c.data.type.length, 9)
        is_(MyClass.__table__.c.data.type.enum_class, FooEnum)

    @testing.variation(
        "sqltype", ["custom", "base_enum", "specific_enum", "string"]
    )
    @testing.variation("indicate_type_explicitly", [True, False])
    def test_pep586_literal(
        self, decl_base, sqltype: Variation, indicate_type_explicitly
    ):
        """test #9187."""

        global Status

        Status = Literal["to-do", "in-progress", "done"]

        if sqltype.custom:

            class LiteralSqlType(types.TypeDecorator):
                impl = types.String
                cache_ok = True

                def __init__(self, literal_type: Any) -> None:
                    super().__init__()
                    self._possible_values = get_args(literal_type)

            our_type = mapped_col_type = LiteralSqlType(Status)
        elif sqltype.specific_enum:
            our_type = mapped_col_type = Enum(
                "to-do", "in-progress", "done", native_enum=False
            )
        elif sqltype.base_enum:
            our_type = Enum(enum.Enum, native_enum=False)
            mapped_col_type = Enum(
                "to-do", "in-progress", "done", native_enum=False
            )
        elif sqltype.string:
            our_type = mapped_col_type = String(50)
        else:
            sqltype.fail()

        decl_base.registry.update_type_annotation_map({Status: our_type})

        class Foo(decl_base):
            __tablename__ = "footable"

            id: Mapped[int] = mapped_column(primary_key=True)

            if indicate_type_explicitly:
                status: Mapped[Status] = mapped_column(mapped_col_type)
            else:
                status: Mapped[Status]

        is_true(isinstance(Foo.__table__.c.status.type, type(our_type)))

        if sqltype.custom:
            eq_(
                Foo.__table__.c.status.type._possible_values,
                ("to-do", "in-progress", "done"),
            )
        elif sqltype.specific_enum or sqltype.base_enum:
            eq_(
                Foo.__table__.c.status.type.enums,
                ["to-do", "in-progress", "done"],
            )
            is_(Foo.__table__.c.status.type.native_enum, False)

    @testing.variation("indicate_type_explicitly", [True, False])
    def test_pep586_literal_defaults_to_enum(
        self, decl_base, indicate_type_explicitly
    ):
        """test #9187."""

        global Status

        Status = Literal["to-do", "in-progress", "done"]

        if indicate_type_explicitly:
            expected_native_enum = True
        else:
            expected_native_enum = False

        class Foo(decl_base):
            __tablename__ = "footable"

            id: Mapped[int] = mapped_column(primary_key=True)

            if indicate_type_explicitly:
                status: Mapped[Status] = mapped_column(
                    Enum("to-do", "in-progress", "done")
                )
            else:
                status: Mapped[Status]

        is_true(isinstance(Foo.__table__.c.status.type, Enum))

        eq_(
            Foo.__table__.c.status.type.enums,
            ["to-do", "in-progress", "done"],
        )
        is_(Foo.__table__.c.status.type.native_enum, expected_native_enum)

    @testing.variation("override_in_type_map", [True, False])
    @testing.variation("indicate_type_explicitly", [True, False])
    def test_pep586_literal_checks_the_arguments(
        self, decl_base, indicate_type_explicitly, override_in_type_map
    ):
        """test #9187."""

        global NotReallyStrings

        NotReallyStrings = Literal["str1", 17, False]

        if override_in_type_map:
            decl_base.registry.update_type_annotation_map(
                {NotReallyStrings: JSON}
            )

        if not override_in_type_map and not indicate_type_explicitly:
            with expect_raises_message(
                ArgumentError,
                "Can't create string-based Enum datatype from non-string "
                "values: 17, False.  Please provide an explicit Enum "
                "datatype for this Python type",
            ):

                class Foo(decl_base):
                    __tablename__ = "footable"

                    id: Mapped[int] = mapped_column(primary_key=True)
                    status: Mapped[NotReallyStrings]

        else:
            # if we override the type in the type_map or mapped_column,
            # then we can again use a Literal with non-strings
            class Foo(decl_base):
                __tablename__ = "footable"

                id: Mapped[int] = mapped_column(primary_key=True)

                if indicate_type_explicitly:
                    status: Mapped[NotReallyStrings] = mapped_column(JSON)
                else:
                    status: Mapped[NotReallyStrings]

            is_true(isinstance(Foo.__table__.c.status.type, JSON))


class MixinTest(fixtures.TestBase, testing.AssertsCompiledSQL):
    __dialect__ = "default"

    def test_mapped_column_omit_fn(self, decl_base):
        class MixinOne:
            name: Mapped[str]
            x: Mapped[int]
            y: Mapped[int] = mapped_column()

        class A(MixinOne, decl_base):
            __tablename__ = "a"
            id: Mapped[int] = mapped_column(primary_key=True)

        eq_(A.__table__.c.keys(), ["id", "name", "x", "y"])

        self.assert_compile(select(A), "SELECT a.id, a.name, a.x, a.y FROM a")

    def test_mapped_column_omit_fn_fixed_table(self, decl_base):
        class MixinOne:
            name: Mapped[str]
            x: Mapped[int]
            y: Mapped[int]

        a = Table(
            "a",
            decl_base.metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(50), nullable=False),
            Column("data", String(50)),
            Column("x", Integer),
            Column("y", Integer),
        )

        class A(MixinOne, decl_base):
            __table__ = a
            id: Mapped[int]

        self.assert_compile(
            select(A), "SELECT a.id, a.name, a.data, a.x, a.y FROM a"
        )

    def test_mc_duplication_plain(self, decl_base):
        class MixinOne:
            name: Mapped[str] = mapped_column()

        class A(MixinOne, decl_base):
            __tablename__ = "a"
            id: Mapped[int] = mapped_column(primary_key=True)

        class B(MixinOne, decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)

        is_not(A.__table__.c.name, B.__table__.c.name)

    def test_mc_duplication_declared_attr(self, decl_base):
        class MixinOne:
            @declared_attr
            def name(cls) -> Mapped[str]:
                return mapped_column()

        class A(MixinOne, decl_base):
            __tablename__ = "a"
            id: Mapped[int] = mapped_column(primary_key=True)

        class B(MixinOne, decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)

        is_not(A.__table__.c.name, B.__table__.c.name)

    def test_relationship_requires_declared_attr(self, decl_base):
        class Related(decl_base):
            __tablename__ = "related"

            id: Mapped[int] = mapped_column(primary_key=True)

        class HasRelated:
            related_id: Mapped[int] = mapped_column(ForeignKey(Related.id))

            related: Mapped[Related] = relationship()

        with expect_raises_message(
            sa_exc.InvalidRequestError,
            r"Mapper properties \(i.e. deferred,column_property\(\), "
            r"relationship\(\), etc.\) must be declared",
        ):

            class A(HasRelated, decl_base):
                __tablename__ = "a"
                id: Mapped[int] = mapped_column(primary_key=True)

    def test_relationship_duplication_declared_attr(self, decl_base):
        class Related(decl_base):
            __tablename__ = "related"

            id: Mapped[int] = mapped_column(primary_key=True)

        class HasRelated:
            related_id: Mapped[int] = mapped_column(ForeignKey(Related.id))

            @declared_attr
            def related(cls) -> Mapped[Related]:
                return relationship()

        class A(HasRelated, decl_base):
            __tablename__ = "a"
            id: Mapped[int] = mapped_column(primary_key=True)

        class B(HasRelated, decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)

        self.assert_compile(
            select(A).join(A.related),
            "SELECT a.id, a.related_id FROM a "
            "JOIN related ON related.id = a.related_id",
        )
        self.assert_compile(
            select(B).join(B.related),
            "SELECT b.id, b.related_id FROM b "
            "JOIN related ON related.id = b.related_id",
        )


class RelationshipLHSTest(fixtures.TestBase, testing.AssertsCompiledSQL):
    __dialect__ = "default"

    @testing.fixture
    def decl_base(self):
        class Base(DeclarativeBase):
            pass

        yield Base
        Base.registry.dispose()

    def test_no_typing_in_rhs(self, decl_base):
        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            bs = relationship("List['B']")

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(Integer, primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))

        with expect_raises_message(
            sa_exc.InvalidRequestError,
            r"When initializing mapper Mapper\[A\(a\)\], expression "
            r'"relationship\(\"List\[\'B\'\]\"\)\" seems to be using a '
            r"generic class as the argument to relationship\(\); please "
            r"state the generic argument using an annotation, e.g. "
            r'"bs: Mapped\[List\[\'B\'\]\] = relationship\(\)"',
        ):

            decl_base.registry.configure()

    def test_required_no_arg(self, decl_base):
        with expect_raises_message(
            sa_exc.ArgumentError,
            r"Python typing annotation is required for attribute "
            r'"A.bs" when primary '
            r'argument\(s\) for "Relationship" construct are None or '
            r"not present",
        ):

            class A(decl_base):
                __tablename__ = "a"

                id: Mapped[int] = mapped_column(primary_key=True)
                bs = relationship()

    def test_legacy_dataclasses_not_currently_using_annotations(
        self, registry
    ):
        """test if relationship() inspects annotations when using
        the legacy dataclass style.

        As of #8692, we are not looking at any annotations that don't use
        ``Mapped[]``.   dataclass users should use MappedAsDataclass and
        new conventions.

        """

        @registry.mapped
        @dataclasses.dataclass
        class A:
            __tablename__ = "a"
            __sa_dataclass_metadata_key__ = "sa"

            id: Mapped[int] = mapped_column(primary_key=True)
            bs: List["B"] = dataclasses.field(  # noqa: F821
                default_factory=list, metadata={"sa": relationship()}
            )

        @registry.mapped
        @dataclasses.dataclass
        class B:
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)
            a_id = mapped_column(ForeignKey("a.id"))

        with expect_raises_message(
            ArgumentError,
            "relationship 'bs' expects a class or a mapper argument",
        ):
            registry.configure()

    @testing.variation(
        "datatype",
        [
            "typing_sequence",
            ("collections_sequence", testing.requires.python310),
            "typing_mutable_sequence",
            ("collections_mutable_sequence", testing.requires.python310),
        ],
    )
    @testing.variation("include_explicit", [True, False])
    def test_relationship_abstract_cls_error(
        self, decl_base, datatype, include_explicit
    ):
        """test #9100"""

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
            data: Mapped[str]

        if include_explicit:

            class A(decl_base):
                __tablename__ = "a"

                id: Mapped[int] = mapped_column(primary_key=True)

                # note this can be done more succinctly by assigning to
                # an interim type, however making it explicit here
                # allows us to further test de-stringifying of these
                # collection types
                if datatype.typing_sequence:
                    bs: Mapped[typing.Sequence[B]] = relationship(
                        collection_class=list
                    )
                elif datatype.collections_sequence:
                    bs: Mapped[collections.abc.Sequence[B]] = relationship(
                        collection_class=list
                    )
                elif datatype.typing_mutable_sequence:
                    bs: Mapped[typing.MutableSequence[B]] = relationship(
                        collection_class=list
                    )
                elif datatype.collections_mutable_sequence:
                    bs: Mapped[
                        collections.abc.MutableSequence[B]
                    ] = relationship(collection_class=list)
                else:
                    datatype.fail()

            decl_base.registry.configure()
            self.assert_compile(
                select(A).join(A.bs),
                "SELECT a.id FROM a JOIN b ON a.id = b.a_id",
            )
        else:
            with expect_raises_message(
                sa_exc.ArgumentError,
                r"Collection annotation type "
                r".*Sequence.* cannot be "
                r"instantiated; please provide an explicit "
                r"'collection_class' parameter \(e.g. list, set, etc.\) to "
                r"the relationship\(\) function to accompany this annotation",
            ):

                class A(decl_base):
                    __tablename__ = "a"

                    id: Mapped[int] = mapped_column(primary_key=True)

                    if datatype.typing_sequence:
                        bs: Mapped[typing.Sequence[B]] = relationship()
                    elif datatype.collections_sequence:
                        bs: Mapped[
                            collections.abc.Sequence[B]
                        ] = relationship()
                    elif datatype.typing_mutable_sequence:
                        bs: Mapped[typing.MutableSequence[B]] = relationship()
                    elif datatype.collections_mutable_sequence:
                        bs: Mapped[
                            collections.abc.MutableSequence[B]
                        ] = relationship()
                    else:
                        datatype.fail()

                decl_base.registry.configure()

    def test_14_style_anno_accepted_w_allow_unmapped(self):
        """test for #8692"""

        class Base(DeclarativeBase):
            __allow_unmapped__ = True

        class A(Base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: str = Column(String)
            bs: List["B"] = relationship("B", back_populates="a")  # noqa: F821

        class B(Base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
            data: Mapped[str]
            a: A = relationship("A", back_populates="bs")

        Base.registry.configure()

        self.assert_compile(
            select(A).join(A.bs),
            "SELECT a.id, a.data FROM a JOIN b ON a.id = b.a_id",
        )
        self.assert_compile(
            select(B).join(B.a),
            "SELECT b.id, b.a_id, b.data FROM b JOIN a ON a.id = b.a_id",
        )

    @testing.combinations(
        ("not_optional",),
        ("optional",),
        ("optional_fwd_ref",),
        ("union_none",),
        ("pep604", testing.requires.python310),
        argnames="optional_on_m2o",
    )
    def test_basic_bidirectional(self, decl_base, optional_on_m2o):
        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column()
            bs: Mapped[List["B"]] = relationship(  # noqa: F821
                back_populates="a"
            )

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(Integer, primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))

            if optional_on_m2o == "optional":
                a: Mapped[Optional["A"]] = relationship(
                    back_populates="bs", primaryjoin=a_id == A.id
                )
            elif optional_on_m2o == "optional_fwd_ref":
                a: Mapped["Optional[A]"] = relationship(
                    back_populates="bs", primaryjoin=a_id == A.id
                )
            elif optional_on_m2o == "union_none":
                a: Mapped["Union[A, None]"] = relationship(
                    back_populates="bs", primaryjoin=a_id == A.id
                )
            elif optional_on_m2o == "pep604":
                a: Mapped[A | None] = relationship(
                    back_populates="bs", primaryjoin=a_id == A.id
                )
            else:
                a: Mapped["A"] = relationship(
                    back_populates="bs", primaryjoin=a_id == A.id
                )

        a1 = A(data="data")
        b1 = B()
        a1.bs.append(b1)
        is_(a1, b1.a)

    def test_wrong_annotation_type_one(self, decl_base):

        with expect_annotation_syntax_error("A.data"):

            class A(decl_base):
                __tablename__ = "a"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: "B" = relationship()  # type: ignore  # noqa

    def test_wrong_annotation_type_two(self, decl_base):

        with expect_annotation_syntax_error("A.data"):

            class B(decl_base):
                __tablename__ = "b"

                id: Mapped[int] = mapped_column(primary_key=True)

            class A(decl_base):
                __tablename__ = "a"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: B = relationship()  # type: ignore  # noqa

    def test_wrong_annotation_type_three(self, decl_base):

        with expect_annotation_syntax_error("A.data"):

            class B(decl_base):
                __tablename__ = "b"

                id: Mapped[int] = mapped_column(primary_key=True)

            class A(decl_base):
                __tablename__ = "a"

                id: Mapped[int] = mapped_column(primary_key=True)
                data: "List[B]" = relationship()  # type: ignore  # noqa

    def test_collection_class_uselist(self, decl_base):
        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column()
            bs_list: Mapped[List["B"]] = relationship(  # noqa: F821
                viewonly=True
            )
            bs_set: Mapped[Set["B"]] = relationship(  # noqa: F821
                viewonly=True
            )
            bs_list_warg: Mapped[List["B"]] = relationship(  # noqa: F821
                "B", viewonly=True
            )
            bs_set_warg: Mapped[Set["B"]] = relationship(  # noqa: F821
                "B", viewonly=True
            )

            # note this is string annotation
            b_one_to_one: Mapped["B"] = relationship(  # noqa: F821
                viewonly=True
            )

            b_one_to_one_warg: Mapped["B"] = relationship(  # noqa: F821
                "B", viewonly=True
            )

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(Integer, primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))

            a: Mapped["A"] = relationship(viewonly=True)
            a_warg: Mapped["A"] = relationship("A", viewonly=True)

        is_(A.__mapper__.attrs["bs_list"].collection_class, list)
        is_(A.__mapper__.attrs["bs_set"].collection_class, set)
        is_(A.__mapper__.attrs["bs_list_warg"].collection_class, list)
        is_(A.__mapper__.attrs["bs_set_warg"].collection_class, set)
        is_true(A.__mapper__.attrs["bs_list"].uselist)
        is_true(A.__mapper__.attrs["bs_set"].uselist)
        is_true(A.__mapper__.attrs["bs_list_warg"].uselist)
        is_true(A.__mapper__.attrs["bs_set_warg"].uselist)

        is_false(A.__mapper__.attrs["b_one_to_one"].uselist)
        is_false(A.__mapper__.attrs["b_one_to_one_warg"].uselist)

        is_false(B.__mapper__.attrs["a"].uselist)
        is_false(B.__mapper__.attrs["a_warg"].uselist)

    def test_one_to_one_example(self, decl_base: Type[DeclarativeBase]):
        """test example in the relationship docs will derive uselist=False
        correctly"""

        class Parent(decl_base):
            __tablename__ = "parent"

            id: Mapped[int] = mapped_column(primary_key=True)
            child: Mapped["Child"] = relationship(  # noqa: F821
                back_populates="parent"
            )

        class Child(decl_base):
            __tablename__ = "child"

            id: Mapped[int] = mapped_column(primary_key=True)
            parent_id: Mapped[int] = mapped_column(ForeignKey("parent.id"))
            parent: Mapped["Parent"] = relationship(back_populates="child")

        c1 = Child()
        p1 = Parent(child=c1)
        is_(p1.child, c1)
        is_(c1.parent, p1)

    def test_collection_class_dict_no_collection(self, decl_base):
        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column()
            bs: Mapped[Dict[str, "B"]] = relationship()  # noqa: F821

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(Integer, primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
            name: Mapped[str] = mapped_column()

        # this is the old collections message.  it's not great, but at the
        # moment I like that this is what's raised
        with expect_raises_message(
            sa_exc.ArgumentError,
            "Type InstrumentedDict must elect an appender",
        ):
            decl_base.registry.configure()

    def test_collection_class_dict_attr_mapped_collection(self, decl_base):
        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column()

            bs: Mapped[KeyFuncDict[str, "B"]] = relationship(  # noqa: F821
                collection_class=attribute_keyed_dict("name")
            )

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(Integer, primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
            name: Mapped[str] = mapped_column()

        decl_base.registry.configure()

        a1 = A()
        b1 = B(name="foo")

        # collection appender on MappedCollection
        a1.bs.set(b1)

        is_(a1.bs["foo"], b1)

    @testing.combinations(
        "include_relationship",
        "no_relationship",
        argnames="include_relationship",
    )
    @testing.combinations(
        "direct_name", "indirect_name", argnames="indirect_name"
    )
    def test_indirect_name_collection(
        self, decl_base, include_relationship, indirect_name
    ):
        """test #8759"""
        global B_

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(Integer, primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))

        B_ = B

        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column()

            if indirect_name == "indirect_name":
                if include_relationship == "include_relationship":
                    bs: Mapped[List[B_]] = relationship("B")
                else:
                    bs: Mapped[List[B_]] = relationship()
            else:
                if include_relationship == "include_relationship":
                    bs: Mapped[List[B]] = relationship("B")
                else:
                    bs: Mapped[List[B]] = relationship()

        self.assert_compile(
            select(A).join(A.bs),
            "SELECT a.id, a.data FROM a JOIN b ON a.id = b.a_id",
        )

    @testing.combinations(
        "include_relationship",
        "no_relationship",
        argnames="include_relationship",
    )
    @testing.combinations(
        "direct_name", "indirect_name", argnames="indirect_name"
    )
    def test_indirect_name_scalar(
        self, decl_base, include_relationship, indirect_name
    ):
        """test #8759"""
        global A_

        class A(decl_base):
            __tablename__ = "a"

            id: Mapped[int] = mapped_column(primary_key=True)
            data: Mapped[str] = mapped_column()

        A_ = A

        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(Integer, primary_key=True)
            a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))

            if indirect_name == "indirect_name":
                if include_relationship == "include_relationship":
                    a: Mapped[A_] = relationship("A")
                else:
                    a: Mapped[A_] = relationship()
            else:
                if include_relationship == "include_relationship":
                    a: Mapped[A] = relationship("A")
                else:
                    a: Mapped[A] = relationship()

        self.assert_compile(
            select(B).join(B.a),
            "SELECT b.id, b.a_id FROM b JOIN a ON a.id = b.a_id",
        )


class CompositeTest(fixtures.TestBase, testing.AssertsCompiledSQL):
    __dialect__ = "default"

    @testing.fixture
    def dataclass_point_fixture(self, decl_base):
        @dataclasses.dataclass
        class Point:
            x: int
            y: int

        class Edge(decl_base):
            __tablename__ = "edge"
            id: Mapped[int] = mapped_column(primary_key=True)
            graph_id: Mapped[int] = mapped_column(ForeignKey("graph.id"))

            start: Mapped[Point] = composite(
                Point, mapped_column("x1"), mapped_column("y1")
            )

            end: Mapped[Point] = composite(
                Point, mapped_column("x2"), mapped_column("y2")
            )

        class Graph(decl_base):
            __tablename__ = "graph"
            id: Mapped[int] = mapped_column(primary_key=True)

            edges: Mapped[List[Edge]] = relationship()

        decl_base.metadata.create_all(testing.db)
        return Point, Graph, Edge

    def test_composite_setup(self, dataclass_point_fixture):
        Point, Graph, Edge = dataclass_point_fixture

        with fixture_session() as sess:
            sess.add(
                Graph(
                    edges=[
                        Edge(start=Point(1, 2), end=Point(3, 4)),
                        Edge(start=Point(7, 8), end=Point(5, 6)),
                    ]
                )
            )
            sess.commit()

        self.assert_compile(
            select(Edge),
            "SELECT edge.id, edge.graph_id, edge.x1, edge.y1, "
            "edge.x2, edge.y2 FROM edge",
        )

        with fixture_session() as sess:
            g1 = sess.scalar(select(Graph))

            # round trip!
            eq_(g1.edges[0].end, Point(3, 4))

    def test_named_setup(self, decl_base: Type[DeclarativeBase]):
        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()

            address: Mapped[Address] = composite(
                Address, mapped_column(), mapped_column(), mapped_column("zip")
            )

        decl_base.metadata.create_all(testing.db)

        with fixture_session() as sess:
            sess.add(
                User(
                    name="user 1",
                    address=Address("123 anywhere street", "NY", "12345"),
                )
            )
            sess.commit()

        with fixture_session() as sess:
            u1 = sess.scalar(select(User))

            # round trip!
            eq_(u1.address, Address("123 anywhere street", "NY", "12345"))

    def test_annotated_setup(self, decl_base):
        global Address

        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()

            address: Mapped["Address"] = composite(
                mapped_column(), mapped_column(), mapped_column("zip")
            )

        self.assert_compile(
            select(User),
            'SELECT "user".id, "user".name, "user".street, '
            '"user".state, "user".zip FROM "user"',
            dialect="default",
        )

    def test_fwd_ref_plus_no_mapped(self, decl_base):
        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        with expect_annotation_syntax_error("User.address"):

            class User(decl_base):
                __tablename__ = "user"

                id: Mapped[int] = mapped_column(primary_key=True)
                name: Mapped[str] = mapped_column()

                address: "Address" = composite(  # type: ignore
                    mapped_column(), mapped_column(), mapped_column("zip")
                )

    def test_extract_from_pep593(self, decl_base):
        global Address

        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()

            address: Mapped[Annotated[Address, "foo"]] = composite(
                mapped_column(), mapped_column(), mapped_column("zip")
            )

        self.assert_compile(
            select(User),
            'SELECT "user".id, "user".name, "user".street, '
            '"user".state, "user".zip FROM "user"',
            dialect="default",
        )

    def test_cls_not_composite_compliant(self, decl_base):
        global Address

        class Address:
            def __init__(self, street: int, state: str, zip_: str):
                pass

            street: str
            state: str
            zip_: str

        with expect_raises_message(
            ArgumentError,
            r"Composite class column arguments must be "
            r"named unless a dataclass is used",
        ):

            class User(decl_base):
                __tablename__ = "user"

                id: Mapped[int] = mapped_column(primary_key=True)
                name: Mapped[str] = mapped_column()

                address: Mapped[Address] = composite(
                    mapped_column(), mapped_column(), mapped_column("zip")
                )

    def test_fwd_ref_ok_explicit_cls(self, decl_base):
        global Address

        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()

            address: Mapped["Address"] = composite(
                Address, mapped_column(), mapped_column(), mapped_column("zip")
            )

        self.assert_compile(
            select(User),
            'SELECT "user".id, "user".name, "user".street, '
            '"user".state, "user".zip FROM "user"',
        )

    def test_name_cols_by_str(self, decl_base):
        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str]
            street: Mapped[str]
            state: Mapped[str]

            # TODO: this needs to be improved, we should be able to say:
            # zip_: Mapped[str] = mapped_column("zip")
            # and it should assign to "zip_" for the attribute. not working

            zip_: Mapped[str] = mapped_column(name="zip", key="zip_")

            address: Mapped["Address"] = composite(
                Address, "street", "state", "zip_"
            )

        eq_(
            User.__mapper__.attrs["address"].props,
            [
                User.__mapper__.attrs["street"],
                User.__mapper__.attrs["state"],
                User.__mapper__.attrs["zip_"],
            ],
        )
        self.assert_compile(
            select(User),
            'SELECT "user".id, "user".name, "user".street, '
            '"user".state, "user".zip FROM "user"',
        )

    def test_cls_annotated_setup(self, decl_base):
        global Address

        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()

            address: Mapped[Address] = composite(
                mapped_column(), mapped_column(), mapped_column("zip")
            )

        decl_base.metadata.create_all(testing.db)

        with fixture_session() as sess:
            sess.add(
                User(
                    name="user 1",
                    address=Address("123 anywhere street", "NY", "12345"),
                )
            )
            sess.commit()

        with fixture_session() as sess:
            u1 = sess.scalar(select(User))

            # round trip!
            eq_(u1.address, Address("123 anywhere street", "NY", "12345"))

    def test_cls_annotated_no_mapped_cols_setup(self, decl_base):
        global Address

        @dataclasses.dataclass
        class Address:
            street: str
            state: str
            zip_: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()

            address: Mapped[Address] = composite()

        decl_base.metadata.create_all(testing.db)

        with fixture_session() as sess:
            sess.add(
                User(
                    name="user 1",
                    address=Address("123 anywhere street", "NY", "12345"),
                )
            )
            sess.commit()

        with fixture_session() as sess:
            u1 = sess.scalar(select(User))

            # round trip!
            eq_(u1.address, Address("123 anywhere street", "NY", "12345"))

    def test_one_col_setup(self, decl_base):
        @dataclasses.dataclass
        class Address:
            street: str

        class User(decl_base):
            __tablename__ = "user"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()

            address: Mapped[Address] = composite(Address, mapped_column())

        decl_base.metadata.create_all(testing.db)

        with fixture_session() as sess:
            sess.add(
                User(
                    name="user 1",
                    address=Address("123 anywhere street"),
                )
            )
            sess.commit()

        with fixture_session() as sess:
            u1 = sess.scalar(select(User))

            # round trip!
            eq_(u1.address, Address("123 anywhere street"))


class AllYourFavoriteHitsTest(fixtures.TestBase, testing.AssertsCompiledSQL):
    """try a bunch of common mappings using the new style"""

    __dialect__ = "default"

    def test_employee_joined_inh(self, decl_base: Type[DeclarativeBase]):
        global str50, str30, opt_str50
        str50 = Annotated[str, 50]
        str30 = Annotated[str, 30]
        opt_str50 = Optional[str50]

        decl_base.registry.update_type_annotation_map(
            {str50: String(50), str30: String(30)}
        )

        class Company(decl_base):
            __tablename__ = "company"

            company_id: Mapped[int] = mapped_column(Integer, primary_key=True)

            name: Mapped[str50]

            employees: Mapped[Set["Person"]] = relationship()  # noqa: F821

        class Person(decl_base):
            __tablename__ = "person"
            person_id: Mapped[int] = mapped_column(primary_key=True)
            company_id: Mapped[int] = mapped_column(
                ForeignKey("company.company_id")
            )
            name: Mapped[str50]
            type: Mapped[str30] = mapped_column()

            __mapper_args__ = {"polymorphic_on": type}

        class Engineer(Person):
            __tablename__ = "engineer"

            person_id: Mapped[int] = mapped_column(
                ForeignKey("person.person_id"), primary_key=True
            )
            __mapper_args__ = {"polymorphic_identity": "engineer"}

            status: Mapped[str] = mapped_column(String(30))
            engineer_name: Mapped[opt_str50]
            primary_language: Mapped[opt_str50]

        class Manager(Person):
            __tablename__ = "manager"

            person_id: Mapped[int] = mapped_column(
                ForeignKey("person.person_id"), primary_key=True
            )
            status: Mapped[str] = mapped_column(String(30))
            manager_name: Mapped[str50]
            __mapper_args__ = {"polymorphic_identity": "manager"}

        is_(Person.__mapper__.polymorphic_on, Person.__table__.c.type)

        # the SELECT statements here confirm the columns present and their
        # ordering
        self.assert_compile(
            select(Person),
            "SELECT person.person_id, person.company_id, person.name, "
            "person.type FROM person",
        )

        self.assert_compile(
            select(Manager),
            "SELECT manager.person_id, person.person_id AS person_id_1, "
            "person.company_id, person.name, person.type, manager.status, "
            "manager.manager_name FROM person "
            "JOIN manager ON person.person_id = manager.person_id",
        )

        self.assert_compile(
            select(Company).join(Company.employees.of_type(Engineer)),
            "SELECT company.company_id, company.name FROM company JOIN "
            "(person JOIN engineer ON person.person_id = engineer.person_id) "
            "ON company.company_id = person.company_id",
        )

    @testing.variation("anno_type", ["plain", "typemap", "annotated"])
    @testing.variation("inh_type", ["single", "joined"])
    def test_mixin_interp_on_inh(self, decl_base, inh_type, anno_type):

        global anno_col

        if anno_type.typemap:
            anno_col = Annotated[str, 30]

            decl_base.registry.update_type_annotation_map({anno_col: String})

            class Mixin:
                foo: Mapped[anno_col]

        elif anno_type.annotated:
            anno_col = Annotated[str, mapped_column(String)]

            class Mixin:
                foo: Mapped[anno_col]

        else:

            class Mixin:
                foo: Mapped[str]

        class Employee(Mixin, decl_base):
            __tablename__ = "employee"

            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str]
            type: Mapped[str]

            __mapper_args__ = {
                "polymorphic_on": "type",
                "polymorphic_identity": "employee",
            }

        class Manager(Employee):
            if inh_type.joined:
                __tablename__ = "manager"

                id: Mapped[int] = mapped_column(  # noqa: A001
                    ForeignKey("employee.id"), primary_key=True
                )

            manager_data: Mapped[str] = mapped_column(nullable=True)

            __mapper_args__ = {
                "polymorphic_identity": "manager",
            }

        if inh_type.single:
            self.assert_compile(
                select(Manager),
                "SELECT employee.id, employee.name, employee.type, "
                "employee.foo, employee.manager_data FROM employee "
                "WHERE employee.type IN (__[POSTCOMPILE_type_1])",
            )
        elif inh_type.joined:
            self.assert_compile(
                select(Manager),
                "SELECT manager.id, employee.id AS id_1, employee.name, "
                "employee.type, employee.foo, manager.manager_data "
                "FROM employee JOIN manager ON employee.id = manager.id",
            )
        else:
            inh_type.fail()


class WriteOnlyRelationshipTest(fixtures.TestBase):
    def _assertions(self, A, B, lazy):
        is_(A.bs.property.mapper, B.__mapper__)

        is_true(A.bs.property.uselist)
        eq_(A.bs.property.lazy, lazy)

    def test_dynamic(self, decl_base):
        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)
            a_id: Mapped[int] = mapped_column(
                ForeignKey("a.id", ondelete="cascade")
            )

        class A(decl_base):
            __tablename__ = "a"
            id: Mapped[int] = mapped_column(primary_key=True)
            bs: DynamicMapped[B] = relationship()

        self._assertions(A, B, "dynamic")

    def test_write_only(self, decl_base):
        class B(decl_base):
            __tablename__ = "b"
            id: Mapped[int] = mapped_column(primary_key=True)
            a_id: Mapped[int] = mapped_column(
                ForeignKey("a.id", ondelete="cascade")
            )

        class A(decl_base):
            __tablename__ = "a"
            id: Mapped[int] = mapped_column(primary_key=True)
            bs: WriteOnlyMapped[B] = relationship()

        self._assertions(A, B, "write_only")


class GenericMappingQueryTest(AssertsCompiledSQL, fixtures.TestBase):
    """test the Generic support added as part of #8665"""

    __dialect__ = "default"

    @testing.fixture
    def mapping(self):
        global T_Value
        T_Value = TypeVar("T_Value")

        class SomeBaseClass(DeclarativeBase):
            pass

        class GenericSetting(
            MappedAsDataclass, SomeBaseClass, Generic[T_Value]
        ):
            """Represents key value pairs for settings or values"""

            __tablename__ = "xx"

            id: Mapped[int] = mapped_column(
                Integer, primary_key=True, init=False
            )

            key: Mapped[str] = mapped_column(String, init=True)

            value: Mapped[T_Value] = mapped_column(
                MutableDict.as_mutable(JSON),
                init=True,
                default_factory=lambda: {},
            )

        return GenericSetting

    def test_inspect(self, mapping):
        GenericSetting = mapping

        typ = GenericSetting[Dict[str, Any]]
        is_(inspect(typ), GenericSetting.__mapper__)

    def test_select(self, mapping):
        GenericSetting = mapping

        typ = GenericSetting[Dict[str, Any]]
        self.assert_compile(
            select(typ).where(typ.key == "x"),
            "SELECT xx.id, xx.key, xx.value FROM xx WHERE xx.key = :key_1",
        )


class BackendTests(fixtures.TestBase):
    __backend__ = True

    @testing.variation("native_enum", [True, False])
    @testing.variation("include_column", [True, False])
    @testing.variation("python_type", ["enum", "literal"])
    def test_schema_type_actually_works(
        self,
        connection,
        decl_base,
        include_column,
        native_enum,
        python_type: Variation,
    ):
        """test that schema type bindings are set up correctly"""

        global Status

        if python_type.enum:

            class Status(enum.Enum):
                PENDING = "pending"
                RECEIVED = "received"
                COMPLETED = "completed"

            enum_argument = [Status]
            test_value = Status.RECEIVED
        elif python_type.literal:
            Status = Literal[  # type: ignore
                "pending", "received", "completed"
            ]
            enum_argument = ["pending", "received", "completed"]
            test_value = "received"
        else:
            python_type.fail()

        if not include_column and not native_enum:
            decl_base.registry.update_type_annotation_map(
                {
                    enum.Enum: Enum(enum.Enum, native_enum=False),
                    Literal: Enum(enum.Enum, native_enum=False),
                }
            )

        class SomeClass(decl_base):
            __tablename__ = "some_table"

            id: Mapped[int] = mapped_column(primary_key=True)

            if include_column:
                status: Mapped[Status] = mapped_column(
                    Enum(
                        *enum_argument,
                        native_enum=bool(native_enum),
                        name="status",
                    )
                )
            else:
                status: Mapped[Status]

        decl_base.metadata.create_all(connection)

        with Session(connection) as sess:
            sess.add(SomeClass(id=1, status=test_value))
            sess.commit()

            eq_(
                sess.scalars(
                    select(SomeClass.status).where(SomeClass.id == 1)
                ).first(),
                test_value,
            )