--- /dev/null
+.. change::
+ :tags: usecase, ext
+ :tickets: 8878
+
+ Added support for the :func:`.association_proxy` extension function to
+ take part within Python ``dataclasses`` configuration, when using
+ the native dataclasses feature described at
+ :ref:`orm_declarative_native_dataclasses`. Included are attribute-level
+ arguments including :paramref:`.association_proxy.init` and
+ :paramref:`.association_proxy.default_factory`.
+
+ Documentation for association proxy has also been updated to use
+ "Annotated Declarative Table" forms within examples, including type
+ annotations used for :class:`.AssocationProxy` itself.
+
+
+.. change::
+ :tags: bug, typing
+
+ Corrected typing support for the :paramref:`_orm.relationship.secondary`
+ argument which may also accept a callable (lambda) that returns a
+ :class:`.FromClause`.
--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 8880
+
+ Fixed bug in dataclass mapping feature where using plain dataclass fields
+ in a mapping would not create a dataclass with the correct class-level
+ state for those fields, copying the raw ``Field`` object to the class
+ inappropriately after dataclasses itself had replaced the ``Field`` object
+ with the class-level default value.
Consider a many-to-many mapping between two classes, ``User`` and ``Keyword``.
Each ``User`` can have any number of ``Keyword`` objects, and vice-versa
-(the many-to-many pattern is described at :ref:`relationships_many_to_many`)::
-
- from sqlalchemy import Column, ForeignKey, Integer, String, Table
- from sqlalchemy.orm import DeclarativeBase, relationship
+(the many-to-many pattern is described at :ref:`relationships_many_to_many`).
+The example below illustrates this pattern in the same way, with the
+exception of an extra attribute added to the ``User`` class called
+``User.keywords``::
+
+ from __future__ import annotations
+
+ from typing import Final
+
+ from sqlalchemy import Column
+ from sqlalchemy import ForeignKey
+ from sqlalchemy import Integer
+ from sqlalchemy import String
+ from sqlalchemy import Table
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
+ from sqlalchemy.orm import relationship
+ from sqlalchemy.ext.associationproxy import association_proxy
+ from sqlalchemy.ext.associationproxy import AssociationProxy
class Base(DeclarativeBase):
class User(Base):
__tablename__ = "user"
- id = mapped_column(Integer, primary_key=True)
- name = mapped_column(String(64))
- kw = relationship("Keyword", secondary=lambda: user_keyword_table)
+ id: Mapped[int] = mapped_column(primary_key=True)
+ name: Mapped[str] = mapped_column(String(64))
+ kw: Mapped[list[Keyword]] = relationship(secondary=lambda: user_keyword_table)
- def __init__(self, name):
+ def __init__(self, name: str):
self.name = name
+ # proxy the 'keyword' attribute from the 'kw' relationship
+ keywords: AssociationProxy[list[str]] = association_proxy("kw", "keyword")
+
class Keyword(Base):
__tablename__ = "keyword"
- id = mapped_column(Integer, primary_key=True)
- keyword = mapped_column("keyword", String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ keyword: Mapped[str] = mapped_column(String(64))
- def __init__(self, keyword):
+ def __init__(self, keyword: str):
self.keyword = keyword
- user_keyword_table = Table(
+ user_keyword_table: Final[Table] = Table(
"user_keyword",
Base.metadata,
Column("user_id", Integer, ForeignKey("user.id"), primary_key=True),
Column("keyword_id", Integer, ForeignKey("keyword.id"), primary_key=True),
)
-Reading and manipulating the collection of "keyword" strings associated
-with ``User`` requires traversal from each collection element to the ``.keyword``
-attribute, which can be awkward::
+In the above example, :func:`.association_proxy` is applied to the ``User``
+class to produce a "view" of the ``kw`` relationship, which exposes the string
+value of ``.keyword`` associated with each ``Keyword`` object. It also
+creates new ``Keyword`` objects transparently when strings are added to the
+collection::
>>> user = User("jek")
- >>> user.kw.append(Keyword("cheese-inspector"))
- >>> print(user.kw)
- [<__main__.Keyword object at 0x12bf830>]
- >>> print(user.kw[0].keyword)
- cheese-inspector
- >>> print([keyword.keyword for keyword in user.kw])
- ['cheese-inspector']
-
-The ``association_proxy`` is applied to the ``User`` class to produce
-a "view" of the ``kw`` relationship, which only exposes the string
-value of ``.keyword`` associated with each ``Keyword`` object::
-
- from sqlalchemy.ext.associationproxy import association_proxy
-
-
- class User(Base):
- __tablename__ = "user"
- id = mapped_column(Integer, primary_key=True)
- name = mapped_column(String(64))
- kw = relationship("Keyword", secondary=lambda: user_keyword_table)
-
- def __init__(self, name):
- self.name = name
-
- # proxy the 'keyword' attribute from the 'kw' relationship
- keywords = association_proxy("kw", "keyword")
+ >>> user.keywords.append("cheese-inspector")
+ >>> user.keywords.append("snack-ninja")
+ >>> print(user.keywords)
+ ['cheese-inspector', 'snack-ninja']
-We can now reference the ``.keywords`` collection as a listing of strings,
-which is both readable and writable. New ``Keyword`` objects are created
-for us transparently::
+To understand the mechanics of this, first review the behavior of
+``User`` and ``Keyword`` without using the ``.keywords`` association proxy.
+Normally, reading and manipulating the collection of "keyword" strings associated
+with ``User`` requires traversal from each collection element to the ``.keyword``
+attribute, which can be awkward. The example below illustrates the identical
+series of operations applied without using the association proxy::
+ >>> # identical operations without using the association proxy
>>> user = User("jek")
- >>> user.keywords.append("cheese-inspector")
- >>> user.keywords
- ['cheese-inspector']
- >>> user.keywords.append("snack ninja")
- >>> user.kw
- [<__main__.Keyword object at 0x12cdd30>, <__main__.Keyword object at 0x12cde30>]
+ >>> user.kw.append(Keyword("cheese-inspector"))
+ >>> user.kw.append(Keyword("snack-ninja"))
+ >>> print([keyword.keyword for keyword in user.kw])
+ ['cheese-inspector', 'snack-ninja']
The :class:`.AssociationProxy` object produced by the :func:`.association_proxy` function
-is an instance of a `Python descriptor <https://docs.python.org/howto/descriptor.html>`_.
-It is always declared with the user-defined class being mapped, regardless of
-whether Declarative or classical mappings via the :class:`_orm.Mapper` function are used.
+is an instance of a `Python descriptor <https://docs.python.org/howto/descriptor.html>`_,
+and is not considered to be "mapped" by the :class:`.Mapper` in any way. Therefore,
+it's always indicated inline within the class definition of the mapped class,
+regardless of whether Declarative or Imperative mappings are used.
The proxy functions by operating upon the underlying mapped attribute
or collection in response to operations, and changes made via the proxy are immediately
or dictionary is taken into account, so that the proxy should act just like
the underlying collection or attribute does.
+.. _associationproxy_creator:
+
Creation of New Values
----------------------
-When a list append() event (or set add(), dictionary __setitem__(), or scalar
-assignment event) is intercepted by the association proxy, it instantiates a
-new instance of the "intermediary" object using its constructor, passing as a
-single argument the given value. In our example above, an operation like::
+When a list ``append()`` event (or set ``add()``, dictionary ``__setitem__()``,
+or scalar assignment event) is intercepted by the association proxy, it
+instantiates a new instance of the "intermediary" object using its constructor,
+passing as a single argument the given value. In our example above, an
+operation like::
user.keywords.append("cheese-inspector")
user.kw.append(Keyword("cheese-inspector"))
The example works here because we have designed the constructor for ``Keyword``
-to accept a single positional argument, ``keyword``. For those cases where a
+to accept a single positional argument, ``keyword``. For those cases where a
single-argument constructor isn't feasible, the association proxy's creational
-behavior can be customized using the ``creator`` argument, which references a
-callable (i.e. Python function) that will produce a new object instance given the
-singular argument. Below we illustrate this using a lambda as is typical::
+behavior can be customized using the :paramref:`.association_proxy.creator`
+argument, which references a callable (i.e. Python function) that will produce
+a new object instance given the singular argument. Below we illustrate this
+using a lambda as is typical::
class User(Base):
- # ...
+ ...
# use Keyword(keyword=kw) on append() events
- keywords = association_proxy(
+ keywords: AssociationProxy[list[str]] = association_proxy(
"kw", "keyword", creator=lambda kw: Keyword(keyword=kw)
)
collection of ``User`` to the ``.keyword`` attribute present on each
``UserKeywordAssociation``::
- from sqlalchemy import Column, ForeignKey, Integer, String
+ from __future__ import annotations
+
+ from typing import Optional
+
+ from sqlalchemy import ForeignKey
+ from sqlalchemy import String
from sqlalchemy.ext.associationproxy import association_proxy
- from sqlalchemy.orm import DeclarativeBase, relationship
+ from sqlalchemy.ext.associationproxy import AssociationProxy
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
+ from sqlalchemy.orm import relationship
class Base(DeclarativeBase):
class User(Base):
__tablename__ = "user"
- id = mapped_column(Integer, primary_key=True)
- name = mapped_column(String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ name: Mapped[str] = mapped_column(String(64))
- user_keyword_associations = relationship(
- "UserKeywordAssociation",
+ user_keyword_associations: Mapped[list[UserKeywordAssociation]] = relationship(
back_populates="user",
cascade="all, delete-orphan",
)
+
# association proxy of "user_keyword_associations" collection
# to "keyword" attribute
- keywords = association_proxy("user_keyword_associations", "keyword")
+ keywords: AssociationProxy[list[Keyword]] = association_proxy(
+ "user_keyword_associations",
+ "keyword",
+ creator=lambda keyword: UserKeywordAssociation(keyword=Keyword(keyword)),
+ )
- def __init__(self, name):
+ def __init__(self, name: str):
self.name = name
class UserKeywordAssociation(Base):
__tablename__ = "user_keyword"
- user_id = mapped_column(Integer, ForeignKey("user.id"), primary_key=True)
- keyword_id = mapped_column(Integer, ForeignKey("keyword.id"), primary_key=True)
- special_key = mapped_column(String(50))
-
- user = relationship(User, back_populates="user_keyword_associations")
+ user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
+ keyword_id: Mapped[int] = mapped_column(ForeignKey("keyword.id"), primary_key=True)
+ special_key: Mapped[Optional[str]] = mapped_column(String(50))
- # reference to the "Keyword" object
- keyword = relationship("Keyword")
+ user: Mapped[User] = relationship(back_populates="user_keyword_associations")
- def __init__(self, keyword=None, user=None, special_key=None):
- self.user = user
- self.keyword = keyword
- self.special_key = special_key
+ keyword: Mapped[Keyword] = relationship()
class Keyword(Base):
__tablename__ = "keyword"
- id = mapped_column(Integer, primary_key=True)
- keyword = mapped_column("keyword", String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ keyword: Mapped[str] = mapped_column("keyword", String(64))
- def __init__(self, keyword):
+ def __init__(self, keyword: str):
self.keyword = keyword
- def __repr__(self):
- return "Keyword(%s)" % repr(self.keyword)
+ def __repr__(self) -> str:
+ return f"Keyword({self.keyword!r})"
With the above configuration, we can operate upon the ``.keywords`` collection
of each ``User`` object, each of which exposes a collection of ``Keyword``
objects that are obtained from the underlying ``UserKeywordAssociation`` elements::
-
>>> user = User("log")
>>> for kw in (Keyword("new_from_blammo"), Keyword("its_big")):
... user.keywords.append(kw)
a collection of strings, rather than a collection of composed objects.
In this case, each ``.keywords.append()`` operation is equivalent to::
- >>> user.user_keyword_associations.append(UserKeywordAssociation(Keyword("its_heavy")))
+ >>> user.user_keyword_associations.append(
+ ... UserKeywordAssociation(keyword=Keyword("its_heavy"))
+ ... )
The ``UserKeywordAssociation`` object has two attributes that are both
populated within the scope of the ``append()`` operation of the association
proxy; ``.keyword``, which refers to the
-``Keyword` object, and ``.user``, which refers to the ``User``.
+``Keyword`` object, and ``.user``, which refers to the ``User`` object.
The ``.keyword`` attribute is populated first, as the association proxy
generates a new ``UserKeywordAssociation`` object in response to the ``.append()``
operation, assigning the given ``Keyword`` instance to the ``.keyword``
construction, has the effect of appending the new ``UserKeywordAssociation`` to
the ``User.user_keyword_associations`` collection (via the relationship)::
- >>> UserKeywordAssociation(Keyword("its_wood"), user, special_key="my special key")
+ >>> UserKeywordAssociation(
+ ... keyword=Keyword("its_wood"), user=user, special_key="my special key"
+ ... )
The association proxy returns to us a collection of ``Keyword`` objects represented
by all these operations::
- >>> user.keywords
+ >>> print(user.keywords)
[Keyword('new_from_blammo'), Keyword('its_big'), Keyword('its_heavy'), Keyword('its_wood')]
.. _proxying_dictionaries:
argument to the ``User.keywords`` proxy so that these values are assigned appropriately
when new elements are added to the dictionary::
- from sqlalchemy import Column, ForeignKey, Integer, String
+ from __future__ import annotations
+
+ from sqlalchemy import ForeignKey
+ from sqlalchemy import String
from sqlalchemy.ext.associationproxy import association_proxy
- from sqlalchemy.orm import DeclarativeBase, relationship
+ from sqlalchemy.ext.associationproxy import AssociationProxy
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
+ from sqlalchemy.orm import relationship
from sqlalchemy.orm.collections import attribute_keyed_dict
class User(Base):
__tablename__ = "user"
- id = mapped_column(Integer, primary_key=True)
- name = mapped_column(String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ name: Mapped[str] = mapped_column(String(64))
# user/user_keyword_associations relationship, mapping
# user_keyword_associations with a dictionary against "special_key" as key.
- user_keyword_associations = relationship(
- "UserKeywordAssociation",
+ user_keyword_associations: Mapped[dict[str, UserKeywordAssociation]] = relationship(
back_populates="user",
collection_class=attribute_keyed_dict("special_key"),
cascade="all, delete-orphan",
# proxy to 'user_keyword_associations', instantiating
# UserKeywordAssociation assigning the new key to 'special_key',
# values to 'keyword'.
- keywords = association_proxy(
+ keywords: AssociationProxy[dict[str, Keyword]] = association_proxy(
"user_keyword_associations",
"keyword",
creator=lambda k, v: UserKeywordAssociation(special_key=k, keyword=v),
)
- def __init__(self, name):
+ def __init__(self, name: str):
self.name = name
class UserKeywordAssociation(Base):
__tablename__ = "user_keyword"
- user_id = mapped_column(Integer, ForeignKey("user.id"), primary_key=True)
- keyword_id = mapped_column(Integer, ForeignKey("keyword.id"), primary_key=True)
- special_key = mapped_column(String)
+ user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
+ keyword_id: Mapped[int] = mapped_column(ForeignKey("keyword.id"), primary_key=True)
+ special_key: Mapped[str]
- user = relationship(
- User,
+ user: Mapped[User] = relationship(
back_populates="user_keyword_associations",
)
- keyword = relationship("Keyword")
+ keyword: Mapped[Keyword] = relationship()
class Keyword(Base):
__tablename__ = "keyword"
- id = mapped_column(Integer, primary_key=True)
- keyword = mapped_column("keyword", String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ keyword: Mapped[str] = mapped_column(String(64))
- def __init__(self, keyword):
+ def __init__(self, keyword: str):
self.keyword = keyword
- def __repr__(self):
- return "Keyword(%s)" % repr(self.keyword)
+ def __repr__(self) -> str:
+ return f"Keyword({self.keyword!r})"
We illustrate the ``.keywords`` collection as a dictionary, mapping the
``UserKeywordAssociation.special_key`` value to ``Keyword`` objects::
an association proxy on ``User`` that refers to an association proxy
present on ``UserKeywordAssociation``::
- from sqlalchemy import Column, ForeignKey, Integer, String
+ from __future__ import annotations
+
+ from sqlalchemy import ForeignKey
+ from sqlalchemy import String
from sqlalchemy.ext.associationproxy import association_proxy
- from sqlalchemy.orm import DeclarativeBase, relationship
+ from sqlalchemy.ext.associationproxy import AssociationProxy
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
+ from sqlalchemy.orm import relationship
from sqlalchemy.orm.collections import attribute_keyed_dict
class User(Base):
__tablename__ = "user"
- id = mapped_column(Integer, primary_key=True)
- name = mapped_column(String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ name: Mapped[str] = mapped_column(String(64))
- user_keyword_associations = relationship(
- "UserKeywordAssociation",
+ user_keyword_associations: Mapped[dict[str, UserKeywordAssociation]] = relationship(
back_populates="user",
collection_class=attribute_keyed_dict("special_key"),
cascade="all, delete-orphan",
)
# the same 'user_keyword_associations'->'keyword' proxy as in
# the basic dictionary example.
- keywords = association_proxy(
+ keywords: AssociationProxy[dict[str, str]] = association_proxy(
"user_keyword_associations",
"keyword",
creator=lambda k, v: UserKeywordAssociation(special_key=k, keyword=v),
)
- def __init__(self, name):
+ def __init__(self, name: str):
self.name = name
class UserKeywordAssociation(Base):
__tablename__ = "user_keyword"
- user_id = mapped_column(ForeignKey("user.id"), primary_key=True)
- keyword_id = mapped_column(ForeignKey("keyword.id"), primary_key=True)
- special_key = mapped_column(String)
- user = relationship(
- User,
+ user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
+ keyword_id: Mapped[int] = mapped_column(ForeignKey("keyword.id"), primary_key=True)
+ special_key: Mapped[str] = mapped_column(String(64))
+ user: Mapped[User] = relationship(
back_populates="user_keyword_associations",
)
# the relationship to Keyword is now called
# 'kw'
- kw = relationship("Keyword")
+ kw: Mapped[Keyword] = relationship()
# 'keyword' is changed to be a proxy to the
# 'keyword' attribute of 'Keyword'
- keyword = association_proxy("kw", "keyword")
+ keyword: AssociationProxy[dict[str, str]] = association_proxy("kw", "keyword")
class Keyword(Base):
__tablename__ = "keyword"
- id = mapped_column(Integer, primary_key=True)
- keyword = mapped_column("keyword", String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ keyword: Mapped[str] = mapped_column(String(64))
- def __init__(self, keyword):
+ def __init__(self, keyword: str):
self.keyword = keyword
``User.keywords`` is now a dictionary of string to string, where
that refers to a column, as well as an association proxy that refers
to a related object, as in the example mapping below::
+ from __future__ import annotations
from sqlalchemy import Column, ForeignKey, Integer, String
- from sqlalchemy.ext.associationproxy import association_proxy
+ from sqlalchemy.ext.associationproxy import association_proxy, AssociationProxy
from sqlalchemy.orm import DeclarativeBase, relationship
from sqlalchemy.orm.collections import attribute_keyed_dict
+ from sqlalchemy.orm.collections import Mapped
class Base(DeclarativeBase):
class User(Base):
__tablename__ = "user"
- id = Column(Integer, primary_key=True)
- name = Column(String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ name: Mapped[str] = mapped_column(String(64))
- user_keyword_associations = relationship(
- "UserKeywordAssociation",
+ user_keyword_associations: Mapped[UserKeywordAssociation] = relationship(
cascade="all, delete-orphan",
)
# object-targeted association proxy
- keywords = association_proxy(
+ keywords: AssociationProxy[List[Keyword]] = association_proxy(
"user_keyword_associations",
"keyword",
)
# column-targeted association proxy
- special_keys = association_proxy("user_keyword_associations", "special_key")
+ special_keys: AssociationProxy[list[str]] = association_proxy(
+ "user_keyword_associations", "special_key"
+ )
class UserKeywordAssociation(Base):
__tablename__ = "user_keyword"
- user_id = Column(ForeignKey("user.id"), primary_key=True)
- keyword_id = Column(ForeignKey("keyword.id"), primary_key=True)
- special_key = Column(String)
- keyword = relationship("Keyword")
+ user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
+ keyword_id: Mapped[int] = mapped_column(ForeignKey("keyword.id"), primary_key=True)
+ special_key: Mapped[str] = mapped_column(String(64))
+ keyword: Mapped[Keyword] = relationship()
class Keyword(Base):
__tablename__ = "keyword"
- id = Column(Integer, primary_key=True)
- keyword = Column("keyword", String(64))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ keyword: Mapped[str] = mapped_column(String(64))
The SQL generated takes the form of a correlated subquery against
the EXISTS SQL operator so that it can be used in a WHERE clause without
Given a mapping as::
+ from __future__ import annotations
+ from sqlalchemy import Column, ForeignKey, Integer, String
+ from sqlalchemy.ext.associationproxy import association_proxy, AssociationProxy
+ from sqlalchemy.orm import DeclarativeBase, relationship
+ from sqlalchemy.orm.collections import attribute_keyed_dict
+ from sqlalchemy.orm.collections import Mapped
+
+
+ class Base(DeclarativeBase):
+ pass
+
+
class A(Base):
__tablename__ = "test_a"
- id = mapped_column(Integer, primary_key=True)
- ab = relationship("AB", backref="a", uselist=False)
- b = association_proxy(
+ id: Mapped[int] = mapped_column(primary_key=True)
+ ab: Mapped[AB] = relationship(uselist=False)
+ b: AssociationProxy[B] = association_proxy(
"ab", "b", creator=lambda b: AB(b=b), cascade_scalar_deletes=True
)
class B(Base):
__tablename__ = "test_b"
- id = mapped_column(Integer, primary_key=True)
- ab = relationship("AB", backref="b", cascade="all, delete-orphan")
+ id: Mapped[int] = mapped_column(primary_key=True)
class AB(Base):
__tablename__ = "test_ab"
- a_id = mapped_column(Integer, ForeignKey(A.id), primary_key=True)
- b_id = mapped_column(Integer, ForeignKey(B.id), primary_key=True)
+ a_id: Mapped[int] = mapped_column(ForeignKey(A.id), primary_key=True)
+ b_id: Mapped[int] = mapped_column(ForeignKey(B.id), primary_key=True)
+
+ b: Mapped[B] = relationship()
An assignment to ``A.b`` will generate an ``AB`` object::
a.b = B()
-The ``A.b`` association is scalar, and includes use of the flag
-:paramref:`.AssociationProxy.cascade_scalar_deletes`. When set, setting ``A.b``
+The ``A.b`` association is scalar, and includes use of the parameter
+:paramref:`.AssociationProxy.cascade_scalar_deletes`. When this parameter
+is enabled, setting ``A.b``
to ``None`` will remove ``A.ab`` as well::
a.b = None
import typing
from typing import AbstractSet
from typing import Any
+from typing import Callable
from typing import cast
from typing import Collection
from typing import Dict
from ..orm import interfaces
from ..orm import ORMDescriptor
from ..orm.base import SQLORMOperations
+from ..orm.interfaces import _AttributeOptions
+from ..orm.interfaces import _DCAttributeOptions
+from ..orm.interfaces import _DEFAULT_ATTRIBUTE_OPTIONS
from ..sql import operators
from ..sql import or_
+from ..sql.base import _NoArg
from ..util.typing import Literal
from ..util.typing import Protocol
from ..util.typing import Self
def association_proxy(
- target_collection: str, attr: str, **kw: Any
+ target_collection: str,
+ attr: str,
+ *,
+ creator: Optional[_CreatorProtocol] = None,
+ getset_factory: Optional[_GetSetFactoryProtocol] = None,
+ proxy_factory: Optional[_ProxyFactoryProtocol] = None,
+ proxy_bulk_set: Optional[_ProxyBulkSetProtocol] = None,
+ info: Optional[_InfoType] = None,
+ cascade_scalar_deletes: bool = False,
+ init: Union[_NoArg, bool] = _NoArg.NO_ARG,
+ repr: Union[_NoArg, bool] = _NoArg.NO_ARG, # noqa: A002
+ default: Optional[Any] = _NoArg.NO_ARG,
+ default_factory: Union[_NoArg, Callable[[], _T]] = _NoArg.NO_ARG,
+ kw_only: Union[_NoArg, bool] = _NoArg.NO_ARG,
) -> AssociationProxy[Any]:
r"""Return a Python property implementing a view of a target
attribute which references an attribute on members of the
the collection type of the target (list, dict or set), or, in the case of
a one to one relationship, a simple scalar value.
- :param target_collection: Name of the attribute we'll proxy to.
- This attribute is typically mapped by
+ :param target_collection: Name of the attribute that is the immediate
+ target. This attribute is typically mapped by
:func:`~sqlalchemy.orm.relationship` to link to a target collection, but
can also be a many-to-one or non-scalar relationship.
- :param attr: Attribute on the associated instance or instances we'll
- proxy for.
+ :param attr: Attribute on the associated instance or instances that
+ are available on instances of the target object.
- For example, given a target collection of [obj1, obj2], a list created
- by this proxy property would look like [getattr(obj1, *attr*),
- getattr(obj2, *attr*)]
+ :param creator: optional.
- If the relationship is one-to-one or otherwise uselist=False, then
- simply: getattr(obj, *attr*)
+ Defines custom behavior when new items are added to the proxied
+ collection.
- :param creator: optional.
+ By default, adding new items to the collection will trigger a
+ construction of an instance of the target object, passing the given
+ item as a positional argument to the target constructor. For cases
+ where this isn't sufficient, :paramref:`.association_proxy.creator`
+ can supply a callable that will construct the object in the
+ appropriate way, given the item that was passed.
+
+ For list- and set- oriented collections, a single argument is
+ passed to the callable. For dictionary oriented collections, two
+ arguments are passed, corresponding to the key and value.
+
+ The :paramref:`.association_proxy.creator` callable is also invoked
+ for scalar (i.e. many-to-one, one-to-one) relationships. If the
+ current value of the target relationship attribute is ``None``, the
+ callable is used to construct a new object. If an object value already
+ exists, the given attribute value is populated onto that object.
+
+ .. seealso::
+
+ :ref:`associationproxy_creator`
+
+ :param cascade_scalar_deletes: when True, indicates that setting
+ the proxied value to ``None``, or deleting it via ``del``, should
+ also remove the source object. Only applies to scalar attributes.
+ Normally, removing the proxied target will not remove the proxy
+ source, as this object may have other state that is still to be
+ kept.
+
+ .. versionadded:: 1.3
+
+ .. seealso::
+
+ :ref:`cascade_scalar_deletes` - complete usage example
+
+ :param init: Specific to :ref:`orm_declarative_native_dataclasses`,
+ specifies if the mapped attribute should be part of the ``__init__()``
+ method as generated by the dataclass process.
+
+ .. versionadded:: 2.0.0b4
+
+ :param repr: Specific to :ref:`orm_declarative_native_dataclasses`,
+ specifies if the attribute established by this :class:`.AssociationProxy`
+ should be part of the ``__repr__()`` method as generated by the dataclass
+ process.
+
+ .. versionadded:: 2.0.0b4
+
+ :param default_factory: Specific to
+ :ref:`orm_declarative_native_dataclasses`, specifies a default-value
+ generation function that will take place as part of the ``__init__()``
+ method as generated by the dataclass process.
+
+ .. versionadded:: 2.0.0b4
+
+ :param kw_only: Specific to :ref:`orm_declarative_native_dataclasses`,
+ indicates if this field should be marked as keyword-only when generating
+ the ``__init__()`` method as generated by the dataclass process.
+
+ .. versionadded:: 2.0.0b4
- When new items are added to this proxied collection, new instances of
- the class collected by the target collection will be created. For list
- and set collections, the target class constructor will be called with
- the 'value' for the new instance. For dict types, two arguments are
- passed: key and value.
+ :param info: optional, will be assigned to
+ :attr:`.AssociationProxy.info` if present.
- If you want to construct instances differently, supply a *creator*
- function that takes arguments as above and returns instances.
- For scalar relationships, creator() will be called if the target is None.
- If the target is present, set operations are proxied to setattr() on the
- associated object.
+ The following additional parameters involve injection of custom behaviors
+ within the :class:`.AssociationProxy` object and are for advanced use
+ only:
- If you have an associated object with multiple attributes, you may set
- up multiple association proxies mapping to different attributes. See
- the unit tests for examples, and for examples of how creator() functions
- can be used to construct the scalar relationship on-demand in this
- situation.
+ :param getset_factory: Optional. Proxied attribute access is
+ automatically handled by routines that get and set values based on
+ the `attr` argument for this proxy.
+
+ If you would like to customize this behavior, you may supply a
+ `getset_factory` callable that produces a tuple of `getter` and
+ `setter` functions. The factory is called with two arguments, the
+ abstract type of the underlying collection and this proxy instance.
+
+ :param proxy_factory: Optional. The type of collection to emulate is
+ determined by sniffing the target collection. If your collection
+ type can't be determined by duck typing or you'd like to use a
+ different collection implementation, you may supply a factory
+ function to produce those collections. Only applicable to
+ non-scalar relationships.
+
+ :param proxy_bulk_set: Optional, use with proxy_factory.
- :param \*\*kw: Passes along any other keyword arguments to
- :class:`.AssociationProxy`.
"""
- return AssociationProxy(target_collection, attr, **kw)
+ return AssociationProxy(
+ target_collection,
+ attr,
+ creator=creator,
+ getset_factory=getset_factory,
+ proxy_factory=proxy_factory,
+ proxy_bulk_set=proxy_bulk_set,
+ info=info,
+ cascade_scalar_deletes=cascade_scalar_deletes,
+ attribute_options=_AttributeOptions(
+ init, repr, default, default_factory, kw_only
+ ),
+ )
class AssociationProxyExtensionType(InspectionAttrExtensionType):
class AssociationProxy(
interfaces.InspectionAttrInfo,
ORMDescriptor[_T],
+ _DCAttributeOptions,
_AssociationProxyProtocol[_T],
):
"""A descriptor that presents a read/write view of an object attribute."""
self,
target_collection: str,
attr: str,
+ *,
creator: Optional[_CreatorProtocol] = None,
getset_factory: Optional[_GetSetFactoryProtocol] = None,
proxy_factory: Optional[_ProxyFactoryProtocol] = None,
proxy_bulk_set: Optional[_ProxyBulkSetProtocol] = None,
info: Optional[_InfoType] = None,
cascade_scalar_deletes: bool = False,
+ attribute_options: Optional[_AttributeOptions] = None,
):
"""Construct a new :class:`.AssociationProxy`.
- The :func:`.association_proxy` function is provided as the usual
- entrypoint here, though :class:`.AssociationProxy` can be instantiated
- and/or subclassed directly.
-
- :param target_collection: Name of the collection we'll proxy to,
- usually created with :func:`_orm.relationship`.
-
- :param attr: Attribute on the collected instances we'll proxy
- for. For example, given a target collection of [obj1, obj2], a
- list created by this proxy property would look like
- [getattr(obj1, attr), getattr(obj2, attr)]
-
- :param creator: Optional. When new items are added to this proxied
- collection, new instances of the class collected by the target
- collection will be created. For list and set collections, the
- target class constructor will be called with the 'value' for the
- new instance. For dict types, two arguments are passed:
- key and value.
-
- If you want to construct instances differently, supply a 'creator'
- function that takes arguments as above and returns instances.
-
- :param cascade_scalar_deletes: when True, indicates that setting
- the proxied value to ``None``, or deleting it via ``del``, should
- also remove the source object. Only applies to scalar attributes.
- Normally, removing the proxied target will not remove the proxy
- source, as this object may have other state that is still to be
- kept.
-
- .. versionadded:: 1.3
-
- .. seealso::
+ The :class:`.AssociationProxy` object is typically constructed using
+ the :func:`.association_proxy` constructor function. See the
+ description of :func:`.association_proxy` for a description of all
+ parameters.
- :ref:`cascade_scalar_deletes` - complete usage example
-
- :param getset_factory: Optional. Proxied attribute access is
- automatically handled by routines that get and set values based on
- the `attr` argument for this proxy.
-
- If you would like to customize this behavior, you may supply a
- `getset_factory` callable that produces a tuple of `getter` and
- `setter` functions. The factory is called with two arguments, the
- abstract type of the underlying collection and this proxy instance.
-
- :param proxy_factory: Optional. The type of collection to emulate is
- determined by sniffing the target collection. If your collection
- type can't be determined by duck typing or you'd like to use a
- different collection implementation, you may supply a factory
- function to produce those collections. Only applicable to
- non-scalar relationships.
-
- :param proxy_bulk_set: Optional, use with proxy_factory. See
- the _set() method for details.
-
- :param info: optional, will be assigned to
- :attr:`.AssociationProxy.info` if present.
-
- .. versionadded:: 1.0.9
"""
self.target_collection = target_collection
if info:
self.info = info # type: ignore
+ if (
+ attribute_options
+ and attribute_options != _DEFAULT_ATTRIBUTE_OPTIONS
+ ):
+ self._has_dataclass_arguments = True
+ self._attribute_options = attribute_options
+ else:
+ self._has_dataclass_arguments = False
+ self._attribute_options = _DEFAULT_ATTRIBUTE_OPTIONS
+
@overload
def __get__(
self: _SelfAssociationProxy, instance: Any, owner: Literal[None]
attributes which are mapped against this type. Below we establish a simple
mapping against the ``my_data`` table::
- from sqlalchemy import mapper
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
- class MyDataClass:
+ class Base(DeclarativeBase):
pass
- # associates mutation listeners with MyDataClass.data
- mapper(MyDataClass, my_data)
+ class MyDataClass(Base):
+ __tablename__ = 'my_data'
+ id: Mapped[int] = mapped_column(primary_key=True)
+ data: Mapped[dict[str, str]] = mapped_column(MutableDict.as_mutable(JSONEncodedDict))
The ``MyDataClass.data`` member will now be notified of in place changes
to its value.
-There's no difference in usage when using declarative::
-
- from sqlalchemy.ext.declarative import declarative_base
-
- Base = declarative_base()
-
- class MyDataClass(Base):
- __tablename__ = 'my_data'
- id = Column(Integer, primary_key=True)
- data = Column(MutableDict.as_mutable(JSONEncodedDict))
-
Any in-place changes to the ``MyDataClass.data`` member
will flag the attribute as "dirty" on the parent object::
>>> from sqlalchemy.orm import Session
- >>> sess = Session()
+ >>> sess = Session(some_engine)
>>> m1 = MyDataClass(data={'value1':'foo'})
>>> sess.add(m1)
>>> sess.commit()
of ``MutableDict`` in all mappings unconditionally, without
the need to declare it individually::
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
+
MutableDict.associate_with(JSONEncodedDict)
+ class Base(DeclarativeBase):
+ pass
+
class MyDataClass(Base):
__tablename__ = 'my_data'
- id = Column(Integer, primary_key=True)
- data = Column(JSONEncodedDict)
+ id: Mapped[int] = mapped_column(primary_key=True)
+ data: Mapped[dict[str, str]] = mapped_column(JSONEncodedDict)
Supporting Pickling
is called when the :func:`.attributes.flag_modified` function is called
from within the mutable extension::
- from sqlalchemy.ext.declarative import declarative_base
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
from sqlalchemy import event
- Base = declarative_base()
+ class Base(DeclarativeBase):
+ pass
class MyDataClass(Base):
__tablename__ = 'my_data'
- id = Column(Integer, primary_key=True)
- data = Column(MutableDict.as_mutable(JSONEncodedDict))
+ id: Mapped[int] = mapped_column(primary_key=True)
+ data: Mapped[dict[str, str]] = mapped_column(MutableDict.as_mutable(JSONEncodedDict))
@event.listens_for(MyDataClass.data, "modified")
def modified_json(instance, initiator):
from .properties import MappedSQLExpression
from .query import AliasOption
from .relationships import _RelationshipArgumentType
+from .relationships import _RelationshipSecondaryArgument
from .relationships import Relationship
from .relationships import RelationshipProperty
from .session import Session
def relationship(
argument: Optional[_RelationshipArgumentType[Any]] = None,
- secondary: Optional[Union[FromClause, str]] = None,
+ secondary: Optional[_RelationshipSecondaryArgument] = None,
*,
uselist: Optional[bool] = None,
collection_class: Optional[
from .descriptor_props import CompositeProperty
from .descriptor_props import SynonymProperty
from .interfaces import _AttributeOptions
+from .interfaces import _DCAttributeOptions
from .interfaces import _IntrospectsAnnotations
from .interfaces import _MappedAttribute
from .interfaces import _MapsColumns
or self.is_dataclass_prior_to_mapping
)
+ look_for_dataclass_things = bool(self.dataclass_setup_arguments)
+
for k in list(collected_attributes):
if k in _include_dunders:
"accidentally placed at the end of the line?" % k
)
continue
- elif not isinstance(value, (Column, MapperProperty, _MapsColumns)):
+ elif look_for_dataclass_things and isinstance(
+ value, dataclasses.Field
+ ):
+ # we collected a dataclass Field; dataclasses would have
+ # set up the correct state on the class
+ continue
+ elif not isinstance(value, (Column, _DCAttributeOptions)):
# using @declared_attr for some object that
- # isn't Column/MapperProperty; remove from the clsdict_view
+ # isn't Column/MapperProperty/_DCAttributeOptions; remove
+ # from the clsdict_view
# and place the evaluated value onto the class.
- if not k.startswith("__"):
- collected_attributes.pop(k)
- self._warn_for_decl_attributes(cls, k, value)
- if not late_mapped:
- setattr(cls, k, value)
+ collected_attributes.pop(k)
+ self._warn_for_decl_attributes(cls, k, value)
+ if not late_mapped:
+ setattr(cls, k, value)
continue
# we expect to see the name 'metadata' in some valid cases;
# however at this point we see it's assigned to something trying
# by util._extract_mapped_subtype before we got here.
assert expect_annotations_wo_mapped
- if (
- isinstance(value, (MapperProperty, _MapsColumns))
- and value._has_dataclass_arguments
- and not self.dataclass_setup_arguments
- ):
- if isinstance(value, MapperProperty):
- argnames = [
- "init",
- "default_factory",
- "repr",
- "default",
- ]
- else:
- argnames = ["init", "default_factory", "repr"]
+ if isinstance(value, _DCAttributeOptions):
+
+ if (
+ value._has_dataclass_arguments
+ and not look_for_dataclass_things
+ ):
+ if isinstance(value, MapperProperty):
+ argnames = [
+ "init",
+ "default_factory",
+ "repr",
+ "default",
+ ]
+ else:
+ argnames = ["init", "default_factory", "repr"]
+
+ args = {
+ a
+ for a in argnames
+ if getattr(
+ value._attribute_options, f"dataclasses_{a}"
+ )
+ is not _NoArg.NO_ARG
+ }
- args = {
- a
- for a in argnames
- if getattr(
- value._attribute_options, f"dataclasses_{a}"
+ raise exc.ArgumentError(
+ f"Attribute '{k}' on class {cls} includes "
+ f"dataclasses argument(s): "
+ f"{', '.join(sorted(repr(a) for a in args))} but "
+ f"class does not specify "
+ "SQLAlchemy native dataclass configuration."
)
- is not _NoArg.NO_ARG
- }
- raise exc.ArgumentError(
- f"Attribute '{k}' on class {cls} includes dataclasses "
- f"argument(s): "
- f"{', '.join(sorted(repr(a) for a in args))} but "
- f"class does not specify "
- "SQLAlchemy native dataclass configuration."
- )
- our_stuff[k] = value
+ if not isinstance(value, (MapperProperty, _MapsColumns)):
+ # filter for _DCAttributeOptions objects that aren't
+ # MapperProperty / mapped_column(). Currently this
+ # includes AssociationProxy. pop it from the things
+ # we're going to map and set it up as a descriptor
+ # on the class.
+ collected_attributes.pop(k)
+
+ # Assoc Prox (or other descriptor object that may
+ # use _DCAttributeOptions) is usually here, except if
+ # 1. we're a
+ # dataclass, dataclasses would have removed the
+ # attr here or 2. assoc proxy is coming from a
+ # superclass, we want it to be direct here so it
+ # tracks state or 3. assoc prox comes from
+ # declared_attr, uncommon case
+ setattr(cls, k, value)
+ continue
+
+ our_stuff[k] = value # type: ignore
def _extract_declared_columns(self) -> None:
our_stuff = self.properties
for this attribute.
"""
- if isinstance(elem, (MapperProperty, _MapsColumns)):
+ if isinstance(elem, _DCAttributeOptions):
dc_field = elem._attribute_options._as_dataclass_field()
return (key, annotation, dc_field)
)
-class _MapsColumns(_MappedAttribute[_T]):
- """interface for declarative-capable construct that delivers one or more
- Column objects to the declarative process to be part of a Table.
+class _DCAttributeOptions:
+ """mixin for descriptors or configurational objects that include dataclass
+ field options.
+
+ This includes :class:`.MapperProperty`, :class:`._MapsColumn` within
+ the ORM, but also includes :class:`.AssociationProxy` within ext.
+ Can in theory be used for other descriptors that serve a similar role
+ as association proxy. (*maybe* hybrids, not sure yet.)
+
"""
__slots__ = ()
_attribute_options: _AttributeOptions
+ """behavioral options for ORM-enabled Python attributes
+
+ .. versionadded:: 2.0
+
+ """
+
_has_dataclass_arguments: bool
+
+class _MapsColumns(_DCAttributeOptions, _MappedAttribute[_T]):
+ """interface for declarative-capable construct that delivers one or more
+ Column objects to the declarative process to be part of a Table.
+ """
+
+ __slots__ = ()
+
@property
def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]:
"""return a MapperProperty to be assigned to the declarative mapping"""
@inspection._self_inspects
class MapperProperty(
HasCacheKey,
+ _DCAttributeOptions,
_MappedAttribute[_T],
InspectionAttrInfo,
util.MemoizedSlots,
doc: Optional[str]
"""optional documentation string"""
- _attribute_options: _AttributeOptions
- """behavioral options for ORM-enabled Python attributes
-
- .. versionadded:: 2.0
-
- """
-
info: _InfoType
"""Info dictionary associated with the object, allowing user-defined
data to be associated with this :class:`.InspectionAttr`.
"""
- _has_dataclass_arguments: bool
-
def _memoized_attr_info(self) -> _InfoType:
"""Info dictionary associated with the object, allowing user-defined
data to be associated with this :class:`.InspectionAttr`.
_RelationshipJoinConditionArgument = Union[
str, _ColumnExpressionArgument[bool]
]
+_RelationshipSecondaryArgument = Union[
+ "FromClause", str, Callable[[], "FromClause"]
+]
_ORMOrderByArgument = Union[
Literal[False],
str,
"""
secondary: _RelationshipArg[
- Optional[Union[FromClause, str]],
+ Optional[_RelationshipSecondaryArgument],
Optional[FromClause],
]
primaryjoin: _RelationshipArg[
def __init__(
self,
argument: Optional[_RelationshipArgumentType[_T]] = None,
- secondary: Optional[Union[FromClause, str]] = None,
+ secondary: Optional[_RelationshipSecondaryArgument] = None,
*,
uselist: Optional[bool] = None,
collection_class: Optional[
--- /dev/null
+from __future__ import annotations
+
+from typing import Final
+
+from sqlalchemy import Column
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy.ext.associationproxy import association_proxy
+from sqlalchemy.ext.associationproxy import AssociationProxy
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
+from sqlalchemy.orm import relationship
+
+
+class Base(DeclarativeBase):
+ pass
+
+
+class User(Base):
+ __tablename__ = "user"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ name: Mapped[str] = mapped_column(String(64))
+ kw: Mapped[list[Keyword]] = relationship(
+ secondary=lambda: user_keyword_table
+ )
+
+ def __init__(self, name: str):
+ self.name = name
+
+ # proxy the 'keyword' attribute from the 'kw' relationship
+ keywords: AssociationProxy[list[str]] = association_proxy("kw", "keyword")
+
+
+class Keyword(Base):
+ __tablename__ = "keyword"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ keyword: Mapped[str] = mapped_column(String(64))
+
+ def __init__(self, keyword: str):
+ self.keyword = keyword
+
+
+user_keyword_table: Final[Table] = Table(
+ "user_keyword",
+ Base.metadata,
+ Column("user_id", Integer, ForeignKey("user.id"), primary_key=True),
+ Column("keyword_id", Integer, ForeignKey("keyword.id"), primary_key=True),
+)
+
+user = User("jek")
+
+# EXPECTED_TYPE: list[Keyword]
+reveal_type(user.kw)
+
+user.kw.append(Keyword("cheese-inspector"))
+
+user.keywords.append("cheese-inspector")
+
+# EXPECTED_TYPE: list[str]
+reveal_type(user.keywords)
+
+user.keywords.append("snack ninja")
+from __future__ import annotations
+
from collections import abc
import copy
+import dataclasses
import pickle
+from typing import List
from unittest.mock import call
from unittest.mock import Mock
from sqlalchemy.engine import default
from sqlalchemy.ext.associationproxy import _AssociationList
from sqlalchemy.ext.associationproxy import association_proxy
+from sqlalchemy.ext.associationproxy import AssociationProxy
from sqlalchemy.orm import aliased
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import collections
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import declared_attr
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import mapper
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
from sqlalchemy.testing.assertions import expect_raises_message
+from sqlalchemy.testing.entities import ComparableMixin # noqa
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
gc_collect()
assert len(a1bs) == 2
+
+
+class DeclOrmForms(fixtures.TestBase):
+ """test issues related to #8880, #8878, #8876"""
+
+ def test_straight_decl_usage(self, decl_base):
+ """test use of assoc prox as the default descriptor for a
+ dataclasses.field.
+
+ """
+
+ class User(decl_base):
+ __allow_unmapped__ = True
+
+ __tablename__ = "user"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ user_keyword_associations: Mapped[
+ List[UserKeywordAssociation]
+ ] = relationship(
+ back_populates="user",
+ cascade="all, delete-orphan",
+ )
+
+ keywords: AssociationProxy[list[str]] = association_proxy(
+ "user_keyword_associations", "keyword"
+ )
+
+ UserKeywordAssociation, Keyword = self._keyword_mapping(
+ User, decl_base
+ )
+
+ self._assert_keyword_assoc_mapping(
+ User, UserKeywordAssociation, Keyword, init=True
+ )
+
+ @testing.variation("embed_in_field", [True, False])
+ @testing.combinations(
+ {},
+ {"repr": False},
+ {"repr": True},
+ ({"kw_only": True}, testing.requires.python310),
+ {"init": False},
+ {"default_factory": True},
+ argnames="field_kw",
+ )
+ def test_dc_decl_usage(self, dc_decl_base, embed_in_field, field_kw):
+ """test use of assoc prox as the default descriptor for a
+ dataclasses.field.
+
+ This exercises #8880
+
+ """
+
+ if field_kw.pop("default_factory", False) and not embed_in_field:
+ has_default_factory = True
+ field_kw["default_factory"] = lambda: [
+ Keyword("l1"),
+ Keyword("l2"),
+ Keyword("l3"),
+ ]
+ else:
+ has_default_factory = False
+
+ class User(dc_decl_base):
+ __allow_unmapped__ = True
+
+ __tablename__ = "user"
+
+ id: Mapped[int] = mapped_column(
+ primary_key=True, repr=True, init=False
+ )
+
+ user_keyword_associations: Mapped[
+ List[UserKeywordAssociation]
+ ] = relationship(
+ back_populates="user",
+ cascade="all, delete-orphan",
+ init=False,
+ )
+
+ if embed_in_field:
+ # this is an incorrect form to use with
+ # MappedAsDataclass. However, we want to make sure it
+ # works as kind of a test to ensure we are being as well
+ # behaved as possible with an explicit dataclasses.field(),
+ # by testing that it uses its normal descriptor-as-default
+ # behavior
+ keywords: AssociationProxy[list[str]] = dataclasses.field(
+ default=association_proxy(
+ "user_keyword_associations", "keyword"
+ ),
+ **field_kw,
+ )
+ else:
+ keywords: AssociationProxy[list[str]] = association_proxy(
+ "user_keyword_associations", "keyword", **field_kw
+ )
+
+ UserKeywordAssociation, Keyword = self._dc_keyword_mapping(
+ User, dc_decl_base
+ )
+
+ # simplify __qualname__ so we can test repr() more easily
+ User.__qualname__ = "mod.User"
+ UserKeywordAssociation.__qualname__ = "mod.UserKeywordAssociation"
+ Keyword.__qualname__ = "mod.Keyword"
+
+ init = field_kw.get("init", True)
+
+ u1 = self._assert_keyword_assoc_mapping(
+ User,
+ UserKeywordAssociation,
+ Keyword,
+ init=init,
+ has_default_factory=has_default_factory,
+ )
+
+ if field_kw.get("repr", True):
+ eq_(
+ repr(u1),
+ "mod.User(id=None, user_keyword_associations=["
+ "mod.UserKeywordAssociation(user_id=None, keyword_id=None, "
+ "keyword=mod.Keyword(id=None, keyword='k1'), user=...), "
+ "mod.UserKeywordAssociation(user_id=None, keyword_id=None, "
+ "keyword=mod.Keyword(id=None, keyword='k2'), user=...), "
+ "mod.UserKeywordAssociation(user_id=None, keyword_id=None, "
+ "keyword=mod.Keyword(id=None, keyword='k3'), user=...)], "
+ "keywords=[mod.Keyword(id=None, keyword='k1'), "
+ "mod.Keyword(id=None, keyword='k2'), "
+ "mod.Keyword(id=None, keyword='k3')])",
+ )
+ else:
+ eq_(
+ repr(u1),
+ "mod.User(id=None, user_keyword_associations=["
+ "mod.UserKeywordAssociation(user_id=None, keyword_id=None, "
+ "keyword=mod.Keyword(id=None, keyword='k1'), user=...), "
+ "mod.UserKeywordAssociation(user_id=None, keyword_id=None, "
+ "keyword=mod.Keyword(id=None, keyword='k2'), user=...), "
+ "mod.UserKeywordAssociation(user_id=None, keyword_id=None, "
+ "keyword=mod.Keyword(id=None, keyword='k3'), user=...)])",
+ )
+
+ def _assert_keyword_assoc_mapping(
+ self,
+ User,
+ UserKeywordAssociation,
+ Keyword,
+ *,
+ init,
+ has_default_factory=False,
+ ):
+ if not init:
+ with expect_raises_message(
+ TypeError, r"got an unexpected keyword argument 'keywords'"
+ ):
+ User(keywords=[Keyword("k1"), Keyword("k2"), Keyword("k3")])
+
+ if has_default_factory:
+ u1 = User()
+ eq_(u1.keywords, [Keyword("l1"), Keyword("l2"), Keyword("l3")])
+
+ eq_(
+ [ka.keyword.keyword for ka in u1.user_keyword_associations],
+ ["l1", "l2", "l3"],
+ )
+
+ if init:
+ u1 = User(keywords=[Keyword("k1"), Keyword("k2"), Keyword("k3")])
+ else:
+ u1 = User()
+ u1.keywords = [Keyword("k1"), Keyword("k2"), Keyword("k3")]
+
+ eq_(u1.keywords, [Keyword("k1"), Keyword("k2"), Keyword("k3")])
+
+ eq_(
+ [ka.keyword.keyword for ka in u1.user_keyword_associations],
+ ["k1", "k2", "k3"],
+ )
+
+ return u1
+
+ def _keyword_mapping(self, User, decl_base):
+ class UserKeywordAssociation(decl_base):
+ __tablename__ = "user_keyword"
+ user_id: Mapped[int] = mapped_column(
+ ForeignKey("user.id"), primary_key=True
+ )
+ keyword_id: Mapped[int] = mapped_column(
+ ForeignKey("keyword.id"), primary_key=True
+ )
+
+ user: Mapped[User] = relationship(
+ back_populates="user_keyword_associations",
+ )
+
+ keyword: Mapped[Keyword] = relationship()
+
+ def __init__(self, keyword=None, user=None):
+ self.user = user
+ self.keyword = keyword
+
+ class Keyword(ComparableMixin, decl_base):
+ __tablename__ = "keyword"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ keyword: Mapped[str] = mapped_column()
+
+ def __init__(self, keyword):
+ self.keyword = keyword
+
+ return UserKeywordAssociation, Keyword
+
+ def _dc_keyword_mapping(self, User, dc_decl_base):
+ class UserKeywordAssociation(dc_decl_base):
+ __tablename__ = "user_keyword"
+ user_id: Mapped[int] = mapped_column(
+ ForeignKey("user.id"), primary_key=True, init=False
+ )
+ keyword_id: Mapped[int] = mapped_column(
+ ForeignKey("keyword.id"), primary_key=True, init=False
+ )
+
+ keyword: Mapped[Keyword] = relationship(default=None)
+
+ user: Mapped[User] = relationship(
+ back_populates="user_keyword_associations", default=None
+ )
+
+ class Keyword(dc_decl_base):
+ __tablename__ = "keyword"
+ id: Mapped[int] = mapped_column(primary_key=True, init=False)
+ keyword: Mapped[str] = mapped_column(init=True)
+
+ return UserKeywordAssociation, Keyword
class B(A):
b_data: Mapped[str] = mapped_column(default="bd")
+ # ensure we didnt break dataclasses contract of removing Field
+ # issue #8880
+ eq_(A.__dict__["some_field"], 5)
+ assert "ctrl_one" not in A.__dict__
+
b1 = B(data="data", ctrl_one="ctrl_one", some_field=5, b_data="x")
eq_(
dataclasses.asdict(b1),
__tablename__ = "test"
id = _column(Integer, primary_key=True)
type_ = _column(String(50))
- __mapper__args = {"polymorphic_on": type_}
+ __mapper_args__ = {"polymorphic_on": type_}
class Specific(General):
__tablename__ = "sub"
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
