--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 10611
+
+ Fixed Declarative issue where typing a relationship using
+ :class:`_orm.Relationship` rather than :class:`_orm.Mapped` would
+ inadvertently pull in the "dynamic" relationship loader strategy for that
+ attribute.
NAMED_TYPE_BUILTINS_LIST = "builtins.list"
NAMED_TYPE_SQLA_MAPPED = "sqlalchemy.orm.base.Mapped"
+_RelFullNames = {
+ "sqlalchemy.orm.relationships.Relationship",
+ "sqlalchemy.orm.relationships.RelationshipProperty",
+ "sqlalchemy.orm.relationships._RelationshipDeclared",
+ "sqlalchemy.orm.Relationship",
+ "sqlalchemy.orm.RelationshipProperty",
+}
+
_lookup: Dict[str, Tuple[int, Set[str]]] = {
"Column": (
COLUMN,
"sqlalchemy.sql.Column",
},
),
- "Relationship": (
- RELATIONSHIP,
- {
- "sqlalchemy.orm.relationships.Relationship",
- "sqlalchemy.orm.relationships.RelationshipProperty",
- "sqlalchemy.orm.Relationship",
- "sqlalchemy.orm.RelationshipProperty",
- },
- ),
- "RelationshipProperty": (
- RELATIONSHIP,
- {
- "sqlalchemy.orm.relationships.Relationship",
- "sqlalchemy.orm.relationships.RelationshipProperty",
- "sqlalchemy.orm.Relationship",
- "sqlalchemy.orm.RelationshipProperty",
- },
- ),
+ "Relationship": (RELATIONSHIP, _RelFullNames),
+ "RelationshipProperty": (RELATIONSHIP, _RelFullNames),
+ "_RelationshipDeclared": (RELATIONSHIP, _RelFullNames),
"registry": (
REGISTRY,
{
from .properties import MappedSQLExpression
from .query import AliasOption
from .relationships import _RelationshipArgumentType
+from .relationships import _RelationshipDeclared
from .relationships import _RelationshipSecondaryArgument
-from .relationships import Relationship
from .relationships import RelationshipProperty
from .session import Session
from .util import _ORMJoin
omit_join: Literal[None, False] = None,
sync_backref: Optional[bool] = None,
**kw: Any,
-) -> Relationship[Any]:
+) -> _RelationshipDeclared[Any]:
"""Provide a relationship between two mapped classes.
This corresponds to a parent-child or associative table relationship.
"""
- return Relationship(
+ return _RelationshipDeclared(
argument,
secondary=secondary,
uselist=uselist,
from typing import no_type_check
from typing import Optional
from typing import overload
+from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
"""
- __slots__ = ()
+ __slots__: Tuple[str, ...] = ()
is_selectable = False
"""Return True if this object is an instance of
from typing import TYPE_CHECKING
from typing import TypeVar
+from .util import _mapper_property_as_plain_name
from .. import exc as sa_exc
from .. import util
from ..exc import MultipleResultsFound # noqa
% (
util.clsname_as_plain_name(actual_strategy_type),
requesting_property,
- util.clsname_as_plain_name(applied_to_property_type),
- util.clsname_as_plain_name(applies_to),
+ _mapper_property_as_plain_name(applied_to_property_type),
+ _mapper_property_as_plain_name(applies_to),
),
)
class _IntrospectsAnnotations:
__slots__ = ()
+ @classmethod
+ def _mapper_property_name(cls) -> str:
+ return cls.__name__
+
def found_in_pep593_annotated(self) -> Any:
"""return a copy of this object to use in declarative when the
object is found inside of an Annotated object."""
raise NotImplementedError(
- f"Use of the {self.__class__} construct inside of an "
- f"Annotated object is not yet supported."
+ f"Use of the {self._mapper_property_name()!r} "
+ "construct inside of an Annotated object is not yet supported."
)
def declarative_scan(
raise sa_exc.ArgumentError(
f"Python typing annotation is required for attribute "
f'"{cls.__name__}.{key}" when primary argument(s) for '
- f'"{self.__class__.__name__}" construct are None or not present'
+ f'"{self._mapper_property_name()}" '
+ "construct are None or not present"
)
argument = extracted_mapped_annotation
assert originating_module is not None
- is_write_only = mapped_container is not None and issubclass(
- mapped_container, WriteOnlyMapped
- )
- if is_write_only:
- self.lazy = "write_only"
- self.strategy_key = (("lazy", self.lazy),)
-
- is_dynamic = mapped_container is not None and issubclass(
- mapped_container, DynamicMapped
- )
- if is_dynamic:
- self.lazy = "dynamic"
- self.strategy_key = (("lazy", self.lazy),)
+ if mapped_container is not None:
+ is_write_only = issubclass(mapped_container, WriteOnlyMapped)
+ is_dynamic = issubclass(mapped_container, DynamicMapped)
+ if is_write_only:
+ self.lazy = "write_only"
+ self.strategy_key = (("lazy", self.lazy),)
+ elif is_dynamic:
+ self.lazy = "dynamic"
+ self.strategy_key = (("lazy", self.lazy),)
+ else:
+ is_write_only = is_dynamic = False
argument = de_optionalize_union_types(argument)
_remote_col_exclude = _ColInAnnotations("remote", "should_not_adapt")
-class Relationship( # type: ignore
+class Relationship(
RelationshipProperty[_T],
_DeclarativeMapped[_T],
- WriteOnlyMapped[_T], # not compatible with Mapped[_T]
- DynamicMapped[_T], # not compatible with Mapped[_T]
):
"""Describes an object property that holds a single item or list
of items that correspond to a related database table.
inherit_cache = True
""":meta private:"""
+
+
+class _RelationshipDeclared( # type: ignore[misc]
+ Relationship[_T],
+ WriteOnlyMapped[_T], # not compatible with Mapped[_T]
+ DynamicMapped[_T], # not compatible with Mapped[_T]
+):
+ """Relationship subclass used implicitly for declarative mapping."""
+
+ inherit_cache = True
+ """:meta private:"""
+
+ @classmethod
+ def _mapper_property_name(cls) -> str:
+ return "Relationship"
)
return annotated.__args__[0], annotated.__origin__
+
+
+def _mapper_property_as_plain_name(prop: Type[Any]) -> str:
+ if hasattr(prop, "_mapper_property_name"):
+ name = prop._mapper_property_name()
+ else:
+ name = None
+ return util.clsname_as_plain_name(prop, name)
return "unprintable element %r" % element
-def clsname_as_plain_name(cls: Type[Any]) -> str:
- return " ".join(
- n.lower() for n in re.findall(r"([A-Z][a-z]+|SQL)", cls.__name__)
- )
+def clsname_as_plain_name(
+ cls: Type[Any], use_name: Optional[str] = None
+) -> str:
+ name = use_name or cls.__name__
+ return " ".join(n.lower() for n in re.findall(r"([A-Z][a-z]+|SQL)", name))
def method_is_overridden(
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 relationship
from sqlalchemy.orm import remote
from sqlalchemy.orm import Session
from sqlalchemy.orm import undefer
from sqlalchemy.orm import WriteOnlyMapped
+from sqlalchemy.orm.attributes import CollectionAttributeImpl
from sqlalchemy.orm.collections import attribute_keyed_dict
from sqlalchemy.orm.collections import KeyFuncDict
+from sqlalchemy.orm.dynamic import DynamicAttributeImpl
from sqlalchemy.orm.properties import MappedColumn
+from sqlalchemy.orm.writeonly import WriteOnlyAttributeImpl
from sqlalchemy.schema import CreateTable
from sqlalchemy.sql.base import _NoArg
from sqlalchemy.sql.sqltypes import Enum
with expect_raises_message(
NotImplementedError,
- r"Use of the \<class 'sqlalchemy.orm."
- r"relationships.Relationship'\> construct inside of an Annotated "
+ r"Use of the 'Relationship' construct inside of an Annotated "
r"object is not yet supported.",
):
yield Base
Base.registry.dispose()
+ @testing.combinations(
+ (Relationship, CollectionAttributeImpl),
+ (Mapped, CollectionAttributeImpl),
+ (WriteOnlyMapped, WriteOnlyAttributeImpl),
+ (DynamicMapped, DynamicAttributeImpl),
+ argnames="mapped_cls,implcls",
+ )
+ def test_use_relationship(self, decl_base, mapped_cls, implcls):
+ """test #10611"""
+
+ 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"))
+
+ class A(decl_base):
+ __tablename__ = "a"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ # for future annotations support, need to write these
+ # directly in source code
+ if mapped_cls is Relationship:
+ bs: Relationship[List[B]] = relationship()
+ elif mapped_cls is Mapped:
+ bs: Mapped[List[B]] = relationship()
+ elif mapped_cls is WriteOnlyMapped:
+ bs: WriteOnlyMapped[List[B]] = relationship()
+ elif mapped_cls is DynamicMapped:
+ bs: DynamicMapped[List[B]] = relationship()
+
+ decl_base.registry.configure()
+ assert isinstance(A.bs.impl, implcls)
+
def test_no_typing_in_rhs(self, decl_base):
class A(decl_base):
__tablename__ = "a"
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 relationship
from sqlalchemy.orm import remote
from sqlalchemy.orm import Session
from sqlalchemy.orm import undefer
from sqlalchemy.orm import WriteOnlyMapped
+from sqlalchemy.orm.attributes import CollectionAttributeImpl
from sqlalchemy.orm.collections import attribute_keyed_dict
from sqlalchemy.orm.collections import KeyFuncDict
+from sqlalchemy.orm.dynamic import DynamicAttributeImpl
from sqlalchemy.orm.properties import MappedColumn
+from sqlalchemy.orm.writeonly import WriteOnlyAttributeImpl
from sqlalchemy.schema import CreateTable
from sqlalchemy.sql.base import _NoArg
from sqlalchemy.sql.sqltypes import Enum
with expect_raises_message(
NotImplementedError,
- r"Use of the \<class 'sqlalchemy.orm."
- r"relationships.Relationship'\> construct inside of an Annotated "
+ r"Use of the 'Relationship' construct inside of an Annotated "
r"object is not yet supported.",
):
yield Base
Base.registry.dispose()
+ @testing.combinations(
+ (Relationship, CollectionAttributeImpl),
+ (Mapped, CollectionAttributeImpl),
+ (WriteOnlyMapped, WriteOnlyAttributeImpl),
+ (DynamicMapped, DynamicAttributeImpl),
+ argnames="mapped_cls,implcls",
+ )
+ def test_use_relationship(self, decl_base, mapped_cls, implcls):
+ """test #10611"""
+
+ # anno only: 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"))
+
+ class A(decl_base):
+ __tablename__ = "a"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ # for future annotations support, need to write these
+ # directly in source code
+ if mapped_cls is Relationship:
+ bs: Relationship[List[B]] = relationship()
+ elif mapped_cls is Mapped:
+ bs: Mapped[List[B]] = relationship()
+ elif mapped_cls is WriteOnlyMapped:
+ bs: WriteOnlyMapped[List[B]] = relationship()
+ elif mapped_cls is DynamicMapped:
+ bs: DynamicMapped[List[B]] = relationship()
+
+ decl_base.registry.configure()
+ assert isinstance(A.bs.impl, implcls)
+
def test_no_typing_in_rhs(self, decl_base):
class A(decl_base):
__tablename__ = "a"
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import registry
+from sqlalchemy.orm import Relationship
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
pass
+class Group(Base):
+ __tablename__ = "group"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+ name: Mapped[str] = mapped_column()
+
+ addresses_style_one_anno_only: Mapped[List["User"]]
+ addresses_style_two_anno_only: Mapped[Set["User"]]
+
+
class User(Base):
__tablename__ = "user"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
+ group_id = mapped_column(ForeignKey("group.id"))
# this currently doesnt generate an error. not sure how to get the
# overloads to hit this one, nor am i sure i really want to do that
user_style_one: Mapped[User] = relationship()
user_style_two: Mapped["User"] = relationship()
+ rel_style_one: Relationship[List["MoreMail"]] = relationship()
+ # everything works even if using Relationship instead of Mapped
+ # users should use Mapped though
+ rel_style_one_anno_only: Relationship[Set["MoreMail"]]
+
+
+class MoreMail(Base):
+ __tablename__ = "address"
+
+ id = mapped_column(Integer, primary_key=True)
+ aggress_id = mapped_column(ForeignKey("address.id"))
+ email: Mapped[str]
+
class SelfReferential(Base):
"""test for #9150"""
# EXPECTED_RE_TYPE: sqlalchemy.orm.attributes.InstrumentedAttribute\[builtins.set\*?\[relationship.Address\]\]
reveal_type(User.addresses_style_two)
+ # EXPECTED_RE_TYPE: sqlalchemy.*.InstrumentedAttribute\[builtins.list\*?\[relationship.User\]\]
+ reveal_type(Group.addresses_style_one_anno_only)
+
+ # EXPECTED_RE_TYPE: sqlalchemy.orm.attributes.InstrumentedAttribute\[builtins.set\*?\[relationship.User\]\]
+ reveal_type(Group.addresses_style_two_anno_only)
+
+ # EXPECTED_RE_TYPE: sqlalchemy.*.InstrumentedAttribute\[builtins.list\*?\[relationship.MoreMail\]\]
+ reveal_type(Address.rel_style_one)
+
+ # EXPECTED_RE_TYPE: sqlalchemy.*.InstrumentedAttribute\[builtins.set\*?\[relationship.MoreMail\]\]
+ reveal_type(Address.rel_style_one_anno_only)
+
mapper_registry: registry = registry()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
