# to be addressable in subqueries
col.key = col._tq_key_label = key
+ # In the rare case of adding a ColumnProperty before the mapper is fully configured (e.g. deferring a reflected column)
+ if not hasattr(self, "columns") or not hasattr(self, "_props"):
+ return prop
+
self.columns.add(col, key)
for col in prop.columns:
descriptor_props = util.preloaded.orm_descriptor_props
- prop = self._props.get(key)
+ prop = self._props.get(key) if hasattr(self, "_props") else None
if isinstance(prop, properties.ColumnProperty):
return self._reconcile_prop_with_incoming_columns(
User()
assert hasattr(User, "addresses")
assert "addresses" in [p.key for p in m1._polymorphic_properties]
+
+ @testing.variation("property_type", ["Column", "ColumnProperty"])
+ def test_add_property_before_mapping_is_complete(self, property_type, connection):
+ m = MetaData()
+ users = Table(
+ "deferred_users",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("name", String),
+ Column("new_column", String),
+ )
+ User = self.classes.User
+ class UserWithDeferred(User):
+ pass
+
+ from sqlalchemy import event
+ # Listen for the event to be able to retrieve the mapper before completing the mapping process
+ @event.listens_for(UserWithDeferred, "instrument_class", propagate=True)
+ def before_mapper_configured(mapper, class_):
+ assert mapper.configured is False
+
+ col = mapper.local_table.c.new_column
+ if property_type.ColumnProperty:
+ mapper.add_property(col.key, deferred(col, group="deferred_group"))
+ else:
+ mapper.add_property(col.key, col)
+
+ if property_type.ColumnProperty:
+ # Make a deferred group
+ self.mapper(
+ UserWithDeferred,
+ users,
+ properties={
+ "name": deferred(users.c.name, group="deferred_group")
+ }
+ )
+ else:
+ self.mapper(UserWithDeferred, users)
+
+ configure_mappers()
+
+ assert hasattr(UserWithDeferred, "new_column")
+
+ m.create_all(connection)
+ sess = Session(connection)
+ u = UserWithDeferred(id=1, name="testuser", new_column="test_value")
+ sess.add(u)
+ sess.commit()
+ sess.close()
+
+ sess = Session(connection)
+ user = sess.query(UserWithDeferred).filter_by(id=1).first()
+ if property_type.ColumnProperty:
+ assert "id" in user.__dict__
+ assert "new_column" not in user.__dict__
+ assert "name" not in user.__dict__
+
+ eq_(user.new_column, "test_value")
+
+ assert "new_column" in user.__dict__
+ assert "name" in user.__dict__
+ eq_(user.name, "testuser")
+
+
+ else:
+ assert "id" in user.__dict__
+ assert "new_column" in user.__dict__
+ assert "name" in user.__dict__
+ eq_(user.new_column, "test_value")
+ eq_(user.name, "testuser")
+
+ sess.close()
+ event.remove(UserWithDeferred, "instrument_class", before_mapper_configured)
def test_replace_col_prop_w_syn(self):
users, User = self.tables.users, self.classes.User