--- /dev/null
+.. change::
+ :tags: usecase, ext
+ :tickets: 10013
+
+ Added new option to :func:`.association_proxy`
+ :paramref:`.association_proxy.create_on_none_assignment`; when an
+ association proxy which refers to a scalar relationship is assigned the
+ value ``None``, and the referenced object is not present, a new object is
+ created via the creator. This was apparently an undefined behavior in the
+ 1.2 series that was silently removed.
proxy_bulk_set: Optional[_ProxyBulkSetProtocol] = None,
info: Optional[_InfoType] = None,
cascade_scalar_deletes: bool = False,
+ create_on_none_assignment: bool = False,
init: Union[_NoArg, bool] = _NoArg.NO_ARG,
repr: Union[_NoArg, bool] = _NoArg.NO_ARG, # noqa: A002
default: Optional[Any] = _NoArg.NO_ARG,
:ref:`cascade_scalar_deletes` - complete usage example
+ :param create_on_none_assignment: when True, indicates that setting
+ the proxied value to ``None`` should **create** the source object
+ if it does not exist, using the creator. Only applies to scalar
+ attributes. This is mutually exclusive
+ vs. the :paramref:`.assocation_proxy.cascade_scalar_deletes`.
+
+ .. versionadded:: 2.0.18
+
: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.
proxy_bulk_set=proxy_bulk_set,
info=info,
cascade_scalar_deletes=cascade_scalar_deletes,
+ create_on_none_assignment=create_on_none_assignment,
attribute_options=_AttributeOptions(
init, repr, default, default_factory, compare, kw_only
),
key: str
target_collection: str
value_attr: str
+ cascade_scalar_deletes: bool
+ create_on_none_assignment: bool
getset_factory: Optional[_GetSetFactoryProtocol]
proxy_factory: Optional[_ProxyFactoryProtocol]
proxy_bulk_set: Optional[_ProxyBulkSetProtocol]
proxy_bulk_set: Optional[_ProxyBulkSetProtocol] = None,
info: Optional[_InfoType] = None,
cascade_scalar_deletes: bool = False,
+ create_on_none_assignment: bool = False,
attribute_options: Optional[_AttributeOptions] = None,
):
"""Construct a new :class:`.AssociationProxy`.
self.getset_factory = getset_factory
self.proxy_factory = proxy_factory
self.proxy_bulk_set = proxy_bulk_set
+
+ if cascade_scalar_deletes and create_on_none_assignment:
+ raise exc.ArgumentError(
+ "The cascade_scalar_deletes and create_on_none_assignment "
+ "parameters are mutually exclusive."
+ )
self.cascade_scalar_deletes = cascade_scalar_deletes
+ self.create_on_none_assignment = create_on_none_assignment
self.key = "_%s_%s_%s" % (
type(self).__name__,
)
target = getattr(obj, self.target_collection)
if target is None:
- if values is None:
+ if (
+ values is None
+ and not self.parent.create_on_none_assignment
+ ):
return
setattr(obj, self.target_collection, creator(values))
else:
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
+from sqlalchemy.testing import is_none
+from sqlalchemy.testing import is_not_none
from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.entities import ComparableMixin # noqa
from sqlalchemy.testing.fixtures import fixture_session
p2 = Parent("p2")
p2.bar = "quux"
+ def test_scalar_opts_exclusive(self):
+ with expect_raises_message(
+ exc.ArgumentError,
+ "The cascade_scalar_deletes and create_on_none_assignment "
+ "parameters are mutually exclusive.",
+ ):
+ association_proxy(
+ "a",
+ "b",
+ cascade_scalar_deletes=True,
+ create_on_none_assignment=True,
+ )
+
+ @testing.variation("create_on_none", [True, False])
+ @testing.variation("specify_creator", [True, False])
+ def test_create_on_set_none(
+ self, create_on_none, specify_creator, decl_base
+ ):
+ class A(decl_base):
+ __tablename__ = "a"
+ id = mapped_column(Integer, primary_key=True)
+ b_id = mapped_column(ForeignKey("b.id"))
+ b = relationship("B")
+
+ if specify_creator:
+ b_data = association_proxy(
+ "b",
+ "data",
+ create_on_none_assignment=bool(create_on_none),
+ creator=lambda data: B(data=data),
+ )
+ else:
+ b_data = association_proxy(
+ "b", "data", create_on_none_assignment=bool(create_on_none)
+ )
+
+ class B(decl_base):
+ __tablename__ = "b"
+ id = mapped_column(Integer, primary_key=True)
+ data = mapped_column(String)
+
+ def __init__(self, data=None):
+ self.data = data
+
+ a1 = A()
+ is_none(a1.b)
+ a1.b_data = None
+
+ if create_on_none:
+ is_not_none(a1.b)
+ else:
+ is_none(a1.b)
+
+ a1.b_data = "data"
+
+ a1.b_data = None
+ is_not_none(a1.b)
+
@testing.provide_metadata
def test_empty_scalars(self):
metadata = self.metadata
useobject = None
cascade_scalar_deletes = None
uselist = None
+ create_on_none_assignment = False
@classmethod
def setup_classes(cls):
"b",
creator=lambda b: AB(b=b),
cascade_scalar_deletes=cls.cascade_scalar_deletes,
+ create_on_none_assignment=cls.create_on_none_assignment,
)
if cls.useobject:
a1.b = None
- assert a1.ab is None
+ if self.create_on_none_assignment:
+ assert isinstance(a1.ab, AB)
+ assert a1.ab is not None
+ eq_(a1.ab.b, None)
+ else:
+ assert a1.ab is None
def test_del_already_nonpresent(self):
if self.useobject:
uselist = False
+class ScalarRemoveScalarObjectNoCascadeNoneAssign(
+ ScalarRemoveScalarObjectNoCascade
+):
+ create_on_none_assignment = True
+
+
class ScalarRemoveListScalarNoCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
uselist = True
+class ScalarRemoveListScalarNoCascadeNoneAssign(
+ ScalarRemoveScalarObjectNoCascade
+):
+ create_on_none_assignment = True
+
+
class ScalarRemoveScalarScalarNoCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
