diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2023-05-10 22:03:18 -0400 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2023-05-10 22:05:45 -0400 |
commit | 73a273c90cda2369ec071435edd9c6dc5c1d31c4 (patch) | |
tree | 27a79e1c0800fffb6a1bce70830b779c0de16251 | |
parent | e00591ec27f63d9cc851bbb3cf4824bd5644a8b8 (diff) | |
download | sqlalchemy-73a273c90cda2369ec071435edd9c6dc5c1d31c4.tar.gz |
return empty MappedColumn() at early pep-593 step
Fixed issue in new ORM Annotated Declarative where using a
:class:`_schema.ForeignKey` (or other column-level constraint) inside of
:func:`_orm.mapped_column` which is then copied out to models via pep-593
``Annotated`` would apply duplicates of each constraint to the
:class:`_schema.Column` as produced in the target :class:`_schema.Table`,
leading to incorrect CREATE TABLE DDL as well as migration directives under
Alembic.
Fixes: #9766
Change-Id: I8a3b2716bf393d1d2b5894f9f72b45fa59df1e08
-rw-r--r-- | doc/build/changelog/unreleased_20/9766.rst | 11 | ||||
-rw-r--r-- | lib/sqlalchemy/orm/properties.py | 10 | ||||
-rw-r--r-- | test/orm/declarative/test_tm_future_annotations_sync.py | 54 | ||||
-rw-r--r-- | test/orm/declarative/test_typed_mapping.py | 54 |
4 files changed, 104 insertions, 25 deletions
diff --git a/doc/build/changelog/unreleased_20/9766.rst b/doc/build/changelog/unreleased_20/9766.rst new file mode 100644 index 000000000..588ce1791 --- /dev/null +++ b/doc/build/changelog/unreleased_20/9766.rst @@ -0,0 +1,11 @@ +.. change:: + :tags: bug, orm + :tickets: 9766 + + Fixed issue in new ORM Annotated Declarative where using a + :class:`_schema.ForeignKey` (or other column-level constraint) inside of + :func:`_orm.mapped_column` which is then copied out to models via pep-593 + ``Annotated`` would apply duplicates of each constraint to the + :class:`_schema.Column` as produced in the target :class:`_schema.Table`, + leading to incorrect CREATE TABLE DDL as well as migration directives under + Alembic. diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py index 916b9d901..dd53a9536 100644 --- a/lib/sqlalchemy/orm/properties.py +++ b/lib/sqlalchemy/orm/properties.py @@ -592,7 +592,6 @@ class MappedColumn( None, SchemaConst.NULL_UNSPECIFIED, ) - util.set_creation_order(self) def _copy(self, **kw: Any) -> Self: @@ -649,7 +648,9 @@ class MappedColumn( return op(col._bind_param(op, other), col, **kwargs) # type: ignore[return-value] # noqa: E501 def found_in_pep593_annotated(self) -> Any: - return self._copy() + # return a blank mapped_column(). This mapped_column()'s + # Column will be merged into it in _init_column_for_annotation(). + return MappedColumn() def declarative_scan( self, @@ -751,13 +752,15 @@ class MappedColumn( if is_pep593(our_type): our_type_is_pep593 = True + pep_593_components = typing_get_args(our_type) raw_pep_593_type = pep_593_components[0] if is_optional_union(raw_pep_593_type): + raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type) + nullable = True if not self._has_nullable: self.column.nullable = nullable - raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type) for elem in pep_593_components[1:]: if isinstance(elem, MappedColumn): use_args_from = elem @@ -772,6 +775,7 @@ class MappedColumn( and use_args_from.column.default is not None ): self.column.default = None + use_args_from.column._merge(self.column) sqltype = self.column.type diff --git a/test/orm/declarative/test_tm_future_annotations_sync.py b/test/orm/declarative/test_tm_future_annotations_sync.py index c1eecf970..dd2c1cc7e 100644 --- a/test/orm/declarative/test_tm_future_annotations_sync.py +++ b/test/orm/declarative/test_tm_future_annotations_sync.py @@ -812,8 +812,10 @@ class MappedColumnTest(fixtures.TestBase, testing.AssertsCompiledSQL): is_true(table.c.data_three.nullable) is_true(table.c.data_four.nullable) + @testing.variation("to_assert", ["ddl", "fkcount", "references"]) + @testing.variation("assign_blank", [True, False]) def test_extract_fk_col_from_pep593( - self, decl_base: Type[DeclarativeBase] + self, decl_base: Type[DeclarativeBase], to_assert, assign_blank ): global intpk, element_ref intpk = Annotated[int, mapped_column(primary_key=True)] @@ -828,28 +830,58 @@ class MappedColumnTest(fixtures.TestBase, testing.AssertsCompiledSQL): __tablename__ = "refone" id: Mapped[intpk] - other_id: Mapped[element_ref] + + if assign_blank: + other_id: Mapped[element_ref] = mapped_column() + else: + other_id: Mapped[element_ref] class RefElementTwo(decl_base): __tablename__ = "reftwo" id: Mapped[intpk] - some_id: Mapped[element_ref] + if assign_blank: + some_id: Mapped[element_ref] = mapped_column() + else: + some_id: Mapped[element_ref] assert Element.__table__ is not None assert RefElementOne.__table__ is not None assert RefElementTwo.__table__ is not None - is_true( - RefElementOne.__table__.c.other_id.references( - Element.__table__.c.id + if to_assert.fkcount: + # test #9766 + eq_(len(RefElementOne.__table__.c.other_id.foreign_keys), 1) + eq_(len(RefElementTwo.__table__.c.some_id.foreign_keys), 1) + elif to_assert.references: + is_true( + RefElementOne.__table__.c.other_id.references( + Element.__table__.c.id + ) ) - ) - is_true( - RefElementTwo.__table__.c.some_id.references( - Element.__table__.c.id + is_true( + RefElementTwo.__table__.c.some_id.references( + Element.__table__.c.id + ) ) - ) + + elif to_assert.ddl: + self.assert_compile( + CreateTable(RefElementOne.__table__), + "CREATE TABLE refone " + "(id INTEGER NOT NULL, other_id INTEGER NOT NULL, " + "PRIMARY KEY (id), " + "FOREIGN KEY(other_id) REFERENCES element (id))", + ) + self.assert_compile( + CreateTable(RefElementTwo.__table__), + "CREATE TABLE reftwo " + "(id INTEGER NOT NULL, some_id INTEGER NOT NULL, " + "PRIMARY KEY (id), " + "FOREIGN KEY(some_id) REFERENCES element (id))", + ) + else: + to_assert.fail() @testing.combinations( (collections.abc.Sequence, (str,), testing.requires.python310), diff --git a/test/orm/declarative/test_typed_mapping.py b/test/orm/declarative/test_typed_mapping.py index 514a4335f..bd7fa1667 100644 --- a/test/orm/declarative/test_typed_mapping.py +++ b/test/orm/declarative/test_typed_mapping.py @@ -803,8 +803,10 @@ class MappedColumnTest(fixtures.TestBase, testing.AssertsCompiledSQL): is_true(table.c.data_three.nullable) is_true(table.c.data_four.nullable) + @testing.variation("to_assert", ["ddl", "fkcount", "references"]) + @testing.variation("assign_blank", [True, False]) def test_extract_fk_col_from_pep593( - self, decl_base: Type[DeclarativeBase] + self, decl_base: Type[DeclarativeBase], to_assert, assign_blank ): # anno only: global intpk, element_ref intpk = Annotated[int, mapped_column(primary_key=True)] @@ -819,28 +821,58 @@ class MappedColumnTest(fixtures.TestBase, testing.AssertsCompiledSQL): __tablename__ = "refone" id: Mapped[intpk] - other_id: Mapped[element_ref] + + if assign_blank: + other_id: Mapped[element_ref] = mapped_column() + else: + other_id: Mapped[element_ref] class RefElementTwo(decl_base): __tablename__ = "reftwo" id: Mapped[intpk] - some_id: Mapped[element_ref] + if assign_blank: + some_id: Mapped[element_ref] = mapped_column() + else: + some_id: Mapped[element_ref] assert Element.__table__ is not None assert RefElementOne.__table__ is not None assert RefElementTwo.__table__ is not None - is_true( - RefElementOne.__table__.c.other_id.references( - Element.__table__.c.id + if to_assert.fkcount: + # test #9766 + eq_(len(RefElementOne.__table__.c.other_id.foreign_keys), 1) + eq_(len(RefElementTwo.__table__.c.some_id.foreign_keys), 1) + elif to_assert.references: + is_true( + RefElementOne.__table__.c.other_id.references( + Element.__table__.c.id + ) ) - ) - is_true( - RefElementTwo.__table__.c.some_id.references( - Element.__table__.c.id + is_true( + RefElementTwo.__table__.c.some_id.references( + Element.__table__.c.id + ) ) - ) + + elif to_assert.ddl: + self.assert_compile( + CreateTable(RefElementOne.__table__), + "CREATE TABLE refone " + "(id INTEGER NOT NULL, other_id INTEGER NOT NULL, " + "PRIMARY KEY (id), " + "FOREIGN KEY(other_id) REFERENCES element (id))", + ) + self.assert_compile( + CreateTable(RefElementTwo.__table__), + "CREATE TABLE reftwo " + "(id INTEGER NOT NULL, some_id INTEGER NOT NULL, " + "PRIMARY KEY (id), " + "FOREIGN KEY(some_id) REFERENCES element (id))", + ) + else: + to_assert.fail() @testing.combinations( (collections.abc.Sequence, (str,), testing.requires.python310), |