--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 9099
+
+ Fixed issue where using a pep-593 ``Annotated`` type in the
+ :paramref:`_orm.registry.type_annotation_map` which itself contained a
+ generic plain container or ``collections.abc`` type (e.g. ``list``,
+ ``dict``, ``collections.abc.Sequence``, etc. ) as the target type would
+ produce an internal error when the ORM were trying to interpret the
+ ``Annotated`` instance.
+
+
--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 9100
+
+ Added an error message when a :func:`_orm.relationship` is mapped against
+ an abstract container type, such as ``Mapped[Sequence[B]]``, without
+ providing the :paramref:`_orm.relationship.container_class` parameter which
+ is necessary when the type is abstract. Previously the the abstract
+ container would attempt to be instantiated at a later step and fail.
+
+
import collections
from collections import abc
import dataclasses
+import inspect as _py_inspect
import re
import typing
from typing import Any
arg_origin, abc.Collection
):
if self.collection_class is None:
+ if _py_inspect.isabstract(arg_origin):
+ raise sa_exc.ArgumentError(
+ f"Collection annotation type {arg_origin} cannot "
+ "be instantiated; please provide an explicit "
+ "'collection_class' parameter "
+ "(e.g. list, set, etc.) to the "
+ "relationship() function to accompany this "
+ "annotation"
+ )
+
self.collection_class = arg_origin
+
elif not is_write_only and not is_dynamic:
self.uselist = False
__args__: Tuple[_AnnotationScanType, ...]
__origin__: Type[_T]
- def copy_with(self, params: Tuple[_AnnotationScanType, ...]) -> Type[_T]:
- ...
+ # Python's builtin _GenericAlias has this method, however builtins like
+ # list, dict, etc. do not, even though they have ``__origin__`` and
+ # ``__args__``
+ #
+ # def copy_with(self, params: Tuple[_AnnotationScanType, ...]) -> Type[_T]:
+ # ...
class SupportsKeysAndGetItem(Protocol[_KT, _VT_co]):
for elem in annotation.__args__
)
- return annotation.copy_with(elements)
+ return _copy_generic_annotation_with(annotation, elements)
return annotation # type: ignore
+def _copy_generic_annotation_with(
+ annotation: GenericProtocol[_T], elements: Tuple[_AnnotationScanType, ...]
+) -> Type[_T]:
+ if hasattr(annotation, "copy_with"):
+ # List, Dict, etc. real generics
+ return annotation.copy_with(elements) # type: ignore
+ else:
+ # Python builtins list, dict, etc.
+ return annotation.__origin__[elements] # type: ignore
+
+
def eval_expression(expression: str, module_name: str) -> Any:
try:
base_globals: Dict[str, Any] = sys.modules[module_name].__dict__
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
)
)
+ @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()),
):
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"""
+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
)
)
+ @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()),
):
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"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
