--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 10369, 10046
+
+ Fixed a wide range of :func:`_orm.mapped_column` parameters that were not
+ being transferred when using the :func:`_orm.mapped_column` object inside
+ of a pep-593 ``Annotated`` object, including
+ :paramref:`_orm.mapped_column.sort_order`,
+ :paramref:`_orm.mapped_column.deferred`,
+ :paramref:`_orm.mapped_column.autoincrement`,
+ :paramref:`_orm.mapped_column.system`, :paramref:`_orm.mapped_column.info`
+ etc.
+
+ Additionally, it remains not supported to have dataclass arguments, such as
+ :paramref:`_orm.mapped_column.kw_only`,
+ :paramref:`_orm.mapped_column.default_factory` etc. indicated within the
+ :func:`_orm.mapped_column` received by ``Annotated``, as this is not
+ supported with pep-681 Dataclass Transforms. A warning is now emitted when
+ these parameters are used within ``Annotated`` in this way (and they
+ continue to be ignored).
primary_key: Optional[bool] = False,
deferred: Union[_NoArg, bool] = _NoArg.NO_ARG,
deferred_group: Optional[str] = None,
- deferred_raiseload: bool = False,
+ deferred_raiseload: Optional[bool] = None,
use_existing_column: bool = False,
name: Optional[str] = None,
type_: Optional[_TypeEngineArgument[Any]] = None,
quote: Optional[bool] = None,
system: bool = False,
comment: Optional[str] = None,
- sort_order: int = 0,
+ sort_order: Union[_NoArg, int] = _NoArg.NO_ARG,
**kw: Any,
) -> MappedColumn[Any]:
r"""declare a new ORM-mapped :class:`_schema.Column` construct
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
+from typing import Union
from . import attributes
from . import strategy_options
"_use_existing_column",
)
- deferred: bool
+ deferred: Union[_NoArg, bool]
deferred_raiseload: bool
deferred_group: Optional[str]
self._use_existing_column = kw.pop("use_existing_column", False)
- self._has_dataclass_arguments = False
-
- if attr_opts is not None and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS:
- if attr_opts.dataclasses_default_factory is not _NoArg.NO_ARG:
- self._has_dataclass_arguments = True
-
- elif (
- attr_opts.dataclasses_init is not _NoArg.NO_ARG
- or attr_opts.dataclasses_repr is not _NoArg.NO_ARG
- ):
- self._has_dataclass_arguments = True
+ self._has_dataclass_arguments = (
+ attr_opts is not None
+ and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS
+ and any(
+ attr_opts[i] is not _NoArg.NO_ARG
+ for i, attr in enumerate(attr_opts._fields)
+ if attr != "dataclasses_default"
+ )
+ )
insert_default = kw.pop("insert_default", _NoArg.NO_ARG)
self._has_insert_default = insert_default is not _NoArg.NO_ARG
self.deferred_group = kw.pop("deferred_group", None)
self.deferred_raiseload = kw.pop("deferred_raiseload", None)
self.deferred = kw.pop("deferred", _NoArg.NO_ARG)
- if self.deferred is _NoArg.NO_ARG:
- self.deferred = bool(
- self.deferred_group or self.deferred_raiseload
- )
self.active_history = kw.pop("active_history", False)
- self._sort_order = kw.pop("sort_order", 0)
+
+ self._sort_order = kw.pop("sort_order", _NoArg.NO_ARG)
self.column = cast("Column[_T]", Column(*arg, **kw))
self.foreign_keys = self.column.foreign_keys
self._has_nullable = "nullable" in kw and kw.get("nullable") not in (
@property
def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]:
- if self.deferred or self.active_history:
+ effective_deferred = self.deferred
+ if effective_deferred is _NoArg.NO_ARG:
+ effective_deferred = bool(
+ self.deferred_group or self.deferred_raiseload
+ )
+
+ if effective_deferred or self.active_history:
return ColumnProperty(
self.column,
- deferred=self.deferred,
+ deferred=effective_deferred,
group=self.deferred_group,
raiseload=self.deferred_raiseload,
attribute_options=self._attribute_options,
@property
def columns_to_assign(self) -> List[Tuple[Column[Any], int]]:
- return [(self.column, self._sort_order)]
+ return [
+ (
+ self.column,
+ self._sort_order
+ if self._sort_order is not _NoArg.NO_ARG
+ else 0,
+ )
+ ]
def __clause_element__(self) -> Column[_T]:
return self.column
use_args_from.column._merge(self.column)
sqltype = self.column.type
+ if (
+ use_args_from.deferred is not _NoArg.NO_ARG
+ and self.deferred is _NoArg.NO_ARG
+ ):
+ self.deferred = use_args_from.deferred
+
+ if (
+ use_args_from.deferred_group is not None
+ and self.deferred_group is None
+ ):
+ self.deferred_group = use_args_from.deferred_group
+
+ if (
+ use_args_from.deferred_raiseload is not None
+ and self.deferred_raiseload is None
+ ):
+ self.deferred_raiseload = use_args_from.deferred_raiseload
+
+ if (
+ use_args_from._use_existing_column
+ and not self._use_existing_column
+ ):
+ self._use_existing_column = True
+
+ if use_args_from.active_history:
+ self.active_history = use_args_from.active_history
+
+ if (
+ use_args_from._sort_order is not None
+ and self._sort_order is _NoArg.NO_ARG
+ ):
+ self._sort_order = use_args_from._sort_order
+
+ if (
+ use_args_from.column.key is not None
+ or use_args_from.column.name is not None
+ ):
+ util.warn_deprecated(
+ "Can't use the 'key' or 'name' arguments in "
+ "Annotated with mapped_column(); this will be ignored",
+ "2.0.22",
+ )
+
+ if use_args_from._has_dataclass_arguments:
+ for idx, arg in enumerate(
+ use_args_from._attribute_options._fields
+ ):
+ if (
+ use_args_from._attribute_options[idx]
+ is not _NoArg.NO_ARG
+ ):
+ arg = arg.replace("dataclasses_", "")
+ util.warn_deprecated(
+ f"Argument '{arg}' is a dataclass argument and "
+ "cannot be specified within a mapped_column() "
+ "bundled inside of an Annotated object",
+ "2.0.22",
+ )
+
if sqltype._isnull and not self.column.foreign_keys:
new_sqltype = None
if self.primary_key:
other.primary_key = True
+ if self.autoincrement != "auto" and other.autoincrement == "auto":
+ other.autoincrement = self.autoincrement
+
+ if self.system:
+ other.system = self.system
+
+ if self.info:
+ other.info.update(self.info)
+
type_ = self.type
if not type_._isnull and other.type._isnull:
if isinstance(type_, SchemaEventTarget):
if self.index and not other.index:
other.index = True
+ if self.doc and other.doc is None:
+ other.doc = self.doc
+
+ if self.comment and other.comment is None:
+ other.comment = self.comment
+
if self.unique and not other.unique:
other.unique = True
import datetime
from decimal import Decimal
import enum
+import inspect as _py_inspect
import typing
from typing import Any
+from typing import cast
from typing import ClassVar
from typing import Dict
from typing import Generic
from sqlalchemy import BigInteger
from sqlalchemy import Column
from sqlalchemy import DateTime
+from sqlalchemy import exc
from sqlalchemy import exc as sa_exc
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy.orm import WriteOnlyMapped
from sqlalchemy.orm.collections import attribute_keyed_dict
from sqlalchemy.orm.collections import KeyFuncDict
+from sqlalchemy.orm.properties import MappedColumn
from sqlalchemy.schema import CreateTable
+from sqlalchemy.sql.base import _NoArg
from sqlalchemy.sql.sqltypes import Enum
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_deprecated
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
Tab.non_existent
+_annotated_names_tested = set()
+
+
+def annotated_name_test_cases(*cases, **kw):
+ _annotated_names_tested.update([case[0] for case in cases])
+
+ return testing.combinations_list(cases, **kw)
+
+
class MappedColumnTest(fixtures.TestBase, testing.AssertsCompiledSQL):
__dialect__ = "default"
is_true(MyClass.__table__.c.data_two.nullable)
eq_(MyClass.__table__.c.data_three.type.length, 50)
+ @testing.requires.python310
+ def test_we_got_all_attrs_test_annotated(self):
+ argnames = _py_inspect.getfullargspec(mapped_column)
+ assert _annotated_names_tested.issuperset(argnames.kwonlyargs), (
+ f"annotated attributes were not tested: "
+ f"{set(argnames.kwonlyargs).difference(_annotated_names_tested)}"
+ )
+
+ @annotated_name_test_cases(
+ ("sort_order", 100, lambda sort_order: sort_order == 100),
+ ("nullable", False, lambda column: column.nullable is False),
+ (
+ "active_history",
+ True,
+ lambda column_property: column_property.active_history is True,
+ ),
+ (
+ "deferred",
+ True,
+ lambda column_property: column_property.deferred is True,
+ ),
+ (
+ "deferred",
+ _NoArg.NO_ARG,
+ lambda column_property: column_property is None,
+ ),
+ (
+ "deferred_group",
+ "mygroup",
+ lambda column_property: column_property.deferred is True
+ and column_property.group == "mygroup",
+ ),
+ (
+ "deferred_raiseload",
+ True,
+ lambda column_property: column_property.deferred is True
+ and column_property.raiseload is True,
+ ),
+ (
+ "server_default",
+ "25",
+ lambda column: column.server_default.arg == "25",
+ ),
+ (
+ "server_onupdate",
+ "25",
+ lambda column: column.server_onupdate.arg == "25",
+ ),
+ (
+ "default",
+ 25,
+ lambda column: column.default.arg == 25,
+ ),
+ (
+ "insert_default",
+ 25,
+ lambda column: column.default.arg == 25,
+ ),
+ (
+ "onupdate",
+ 25,
+ lambda column: column.onupdate.arg == 25,
+ ),
+ ("doc", "some doc", lambda column: column.doc == "some doc"),
+ (
+ "comment",
+ "some comment",
+ lambda column: column.comment == "some comment",
+ ),
+ ("index", True, lambda column: column.index is True),
+ ("index", _NoArg.NO_ARG, lambda column: column.index is None),
+ ("unique", True, lambda column: column.unique is True),
+ ("autoincrement", True, lambda column: column.autoincrement is True),
+ ("system", True, lambda column: column.system is True),
+ ("primary_key", True, lambda column: column.primary_key is True),
+ ("type_", BIGINT, lambda column: isinstance(column.type, BIGINT)),
+ ("info", {"foo": "bar"}, lambda column: column.info == {"foo": "bar"}),
+ (
+ "use_existing_column",
+ True,
+ lambda mc: mc._use_existing_column is True,
+ ),
+ (
+ "quote",
+ True,
+ exc.SADeprecationWarning(
+ "Can't use the 'key' or 'name' arguments in Annotated "
+ ),
+ ),
+ (
+ "key",
+ "mykey",
+ exc.SADeprecationWarning(
+ "Can't use the 'key' or 'name' arguments in Annotated "
+ ),
+ ),
+ (
+ "name",
+ "mykey",
+ exc.SADeprecationWarning(
+ "Can't use the 'key' or 'name' arguments in Annotated "
+ ),
+ ),
+ (
+ "kw_only",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'kw_only' is a dataclass argument "
+ ),
+ testing.requires.python310,
+ ),
+ (
+ "compare",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'compare' is a dataclass argument "
+ ),
+ testing.requires.python310,
+ ),
+ (
+ "default_factory",
+ lambda: 25,
+ exc.SADeprecationWarning(
+ "Argument 'default_factory' is a dataclass argument "
+ ),
+ ),
+ (
+ "repr",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'repr' is a dataclass argument "
+ ),
+ ),
+ (
+ "init",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'init' is a dataclass argument"
+ ),
+ ),
+ argnames="argname, argument, assertion",
+ )
+ @testing.variation("use_annotated", [True, False, "control"])
+ def test_names_encountered_for_annotated(
+ self, argname, argument, assertion, use_annotated, decl_base
+ ):
+ global myint
+
+ if argument is not _NoArg.NO_ARG:
+ kw = {argname: argument}
+
+ if argname == "quote":
+ kw["name"] = "somename"
+ else:
+ kw = {}
+
+ is_warning = isinstance(assertion, exc.SADeprecationWarning)
+ is_dataclass = argname in (
+ "kw_only",
+ "init",
+ "repr",
+ "compare",
+ "default_factory",
+ )
+
+ if is_dataclass:
+
+ class Base(MappedAsDataclass, decl_base):
+ __abstract__ = True
+
+ else:
+ Base = decl_base
+
+ if use_annotated.control:
+ # test in reverse; that kw set on the main mapped_column() takes
+ # effect when the Annotated is there also and does not have the
+ # kw
+ amc = mapped_column()
+ myint = Annotated[int, amc]
+
+ mc = mapped_column(**kw)
+
+ class User(Base):
+ __tablename__ = "user"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ myname: Mapped[myint] = mc
+
+ elif use_annotated:
+ amc = mapped_column(**kw)
+ myint = Annotated[int, amc]
+
+ mc = mapped_column()
+
+ if is_warning:
+ with expect_deprecated(assertion.args[0]):
+
+ class User(Base):
+ __tablename__ = "user"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ myname: Mapped[myint] = mc
+
+ else:
+
+ class User(Base):
+ __tablename__ = "user"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ myname: Mapped[myint] = mc
+
+ else:
+ mc = cast(MappedColumn, mapped_column(**kw))
+
+ mapper_prop = mc.mapper_property_to_assign
+ column_to_assign, sort_order = mc.columns_to_assign[0]
+
+ if not is_warning:
+ assert_result = testing.resolve_lambda(
+ assertion,
+ sort_order=sort_order,
+ column_property=mapper_prop,
+ column=column_to_assign,
+ mc=mc,
+ )
+ assert assert_result
+ elif is_dataclass and (not use_annotated or use_annotated.control):
+ eq_(
+ getattr(mc._attribute_options, f"dataclasses_{argname}"),
+ argument,
+ )
+
def test_pep484_newtypes_as_typemap_keys(
self, decl_base: Type[DeclarativeBase]
):
import datetime
from decimal import Decimal
import enum
+import inspect as _py_inspect
import typing
from typing import Any
+from typing import cast
from typing import ClassVar
from typing import Dict
from typing import Generic
from sqlalchemy import BigInteger
from sqlalchemy import Column
from sqlalchemy import DateTime
+from sqlalchemy import exc
from sqlalchemy import exc as sa_exc
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy.orm import WriteOnlyMapped
from sqlalchemy.orm.collections import attribute_keyed_dict
from sqlalchemy.orm.collections import KeyFuncDict
+from sqlalchemy.orm.properties import MappedColumn
from sqlalchemy.schema import CreateTable
+from sqlalchemy.sql.base import _NoArg
from sqlalchemy.sql.sqltypes import Enum
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_deprecated
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
Tab.non_existent
+_annotated_names_tested = set()
+
+
+def annotated_name_test_cases(*cases, **kw):
+ _annotated_names_tested.update([case[0] for case in cases])
+
+ return testing.combinations_list(cases, **kw)
+
+
class MappedColumnTest(fixtures.TestBase, testing.AssertsCompiledSQL):
__dialect__ = "default"
is_true(MyClass.__table__.c.data_two.nullable)
eq_(MyClass.__table__.c.data_three.type.length, 50)
+ @testing.requires.python310
+ def test_we_got_all_attrs_test_annotated(self):
+ argnames = _py_inspect.getfullargspec(mapped_column)
+ assert _annotated_names_tested.issuperset(argnames.kwonlyargs), (
+ f"annotated attributes were not tested: "
+ f"{set(argnames.kwonlyargs).difference(_annotated_names_tested)}"
+ )
+
+ @annotated_name_test_cases(
+ ("sort_order", 100, lambda sort_order: sort_order == 100),
+ ("nullable", False, lambda column: column.nullable is False),
+ (
+ "active_history",
+ True,
+ lambda column_property: column_property.active_history is True,
+ ),
+ (
+ "deferred",
+ True,
+ lambda column_property: column_property.deferred is True,
+ ),
+ (
+ "deferred",
+ _NoArg.NO_ARG,
+ lambda column_property: column_property is None,
+ ),
+ (
+ "deferred_group",
+ "mygroup",
+ lambda column_property: column_property.deferred is True
+ and column_property.group == "mygroup",
+ ),
+ (
+ "deferred_raiseload",
+ True,
+ lambda column_property: column_property.deferred is True
+ and column_property.raiseload is True,
+ ),
+ (
+ "server_default",
+ "25",
+ lambda column: column.server_default.arg == "25",
+ ),
+ (
+ "server_onupdate",
+ "25",
+ lambda column: column.server_onupdate.arg == "25",
+ ),
+ (
+ "default",
+ 25,
+ lambda column: column.default.arg == 25,
+ ),
+ (
+ "insert_default",
+ 25,
+ lambda column: column.default.arg == 25,
+ ),
+ (
+ "onupdate",
+ 25,
+ lambda column: column.onupdate.arg == 25,
+ ),
+ ("doc", "some doc", lambda column: column.doc == "some doc"),
+ (
+ "comment",
+ "some comment",
+ lambda column: column.comment == "some comment",
+ ),
+ ("index", True, lambda column: column.index is True),
+ ("index", _NoArg.NO_ARG, lambda column: column.index is None),
+ ("unique", True, lambda column: column.unique is True),
+ ("autoincrement", True, lambda column: column.autoincrement is True),
+ ("system", True, lambda column: column.system is True),
+ ("primary_key", True, lambda column: column.primary_key is True),
+ ("type_", BIGINT, lambda column: isinstance(column.type, BIGINT)),
+ ("info", {"foo": "bar"}, lambda column: column.info == {"foo": "bar"}),
+ (
+ "use_existing_column",
+ True,
+ lambda mc: mc._use_existing_column is True,
+ ),
+ (
+ "quote",
+ True,
+ exc.SADeprecationWarning(
+ "Can't use the 'key' or 'name' arguments in Annotated "
+ ),
+ ),
+ (
+ "key",
+ "mykey",
+ exc.SADeprecationWarning(
+ "Can't use the 'key' or 'name' arguments in Annotated "
+ ),
+ ),
+ (
+ "name",
+ "mykey",
+ exc.SADeprecationWarning(
+ "Can't use the 'key' or 'name' arguments in Annotated "
+ ),
+ ),
+ (
+ "kw_only",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'kw_only' is a dataclass argument "
+ ),
+ testing.requires.python310,
+ ),
+ (
+ "compare",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'compare' is a dataclass argument "
+ ),
+ testing.requires.python310,
+ ),
+ (
+ "default_factory",
+ lambda: 25,
+ exc.SADeprecationWarning(
+ "Argument 'default_factory' is a dataclass argument "
+ ),
+ ),
+ (
+ "repr",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'repr' is a dataclass argument "
+ ),
+ ),
+ (
+ "init",
+ True,
+ exc.SADeprecationWarning(
+ "Argument 'init' is a dataclass argument"
+ ),
+ ),
+ argnames="argname, argument, assertion",
+ )
+ @testing.variation("use_annotated", [True, False, "control"])
+ def test_names_encountered_for_annotated(
+ self, argname, argument, assertion, use_annotated, decl_base
+ ):
+ # anno only: global myint
+
+ if argument is not _NoArg.NO_ARG:
+ kw = {argname: argument}
+
+ if argname == "quote":
+ kw["name"] = "somename"
+ else:
+ kw = {}
+
+ is_warning = isinstance(assertion, exc.SADeprecationWarning)
+ is_dataclass = argname in (
+ "kw_only",
+ "init",
+ "repr",
+ "compare",
+ "default_factory",
+ )
+
+ if is_dataclass:
+
+ class Base(MappedAsDataclass, decl_base):
+ __abstract__ = True
+
+ else:
+ Base = decl_base
+
+ if use_annotated.control:
+ # test in reverse; that kw set on the main mapped_column() takes
+ # effect when the Annotated is there also and does not have the
+ # kw
+ amc = mapped_column()
+ myint = Annotated[int, amc]
+
+ mc = mapped_column(**kw)
+
+ class User(Base):
+ __tablename__ = "user"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ myname: Mapped[myint] = mc
+
+ elif use_annotated:
+ amc = mapped_column(**kw)
+ myint = Annotated[int, amc]
+
+ mc = mapped_column()
+
+ if is_warning:
+ with expect_deprecated(assertion.args[0]):
+
+ class User(Base):
+ __tablename__ = "user"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ myname: Mapped[myint] = mc
+
+ else:
+
+ class User(Base):
+ __tablename__ = "user"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ myname: Mapped[myint] = mc
+
+ else:
+ mc = cast(MappedColumn, mapped_column(**kw))
+
+ mapper_prop = mc.mapper_property_to_assign
+ column_to_assign, sort_order = mc.columns_to_assign[0]
+
+ if not is_warning:
+ assert_result = testing.resolve_lambda(
+ assertion,
+ sort_order=sort_order,
+ column_property=mapper_prop,
+ column=column_to_assign,
+ mc=mc,
+ )
+ assert assert_result
+ elif is_dataclass and (not use_annotated or use_annotated.control):
+ eq_(
+ getattr(mc._attribute_options, f"dataclasses_{argname}"),
+ argument,
+ )
+
def test_pep484_newtypes_as_typemap_keys(
self, decl_base: Type[DeclarativeBase]
):
("unique", True),
("type", BigInteger()),
("type", Enum("one", "two", "three", create_constraint=True)),
+ ("doc", "some doc"),
+ ("comment", "some comment"),
+ ("system", True),
+ ("autoincrement", True),
+ ("info", {"foo": "bar"}),
argnames="paramname, value",
)
def test_merge_column(
# so here it's getting mutated in place. this is a bug
is_(default.column, target_copy)
+ elif paramname in ("info",):
+ eq_(col.info, value)
elif paramname == "type":
assert type(col.type) is type(value)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
