Я предполагаю, что ваши Base классовые миксы в столбце name?
Ваша цель состоит в том, чтобы inspect(category).committed_state выглядел так, как это выглядит для вновь созданных объектов (за исключением, может быть, атрибута id).То же самое для каждого объекта продукта.
В вашем примере с "вновь созданными объектами" category committed_state выглядит так до сброса сеанса:
{'id': symbol('NEVER_SET'),
'name': symbol('NO_VALUE'),
'products': [],
'top_product': symbol('NEVER_SET')}
, а productcommitted_state выглядит следующим образом:
{'category': symbol('NEVER_SET'),
'id': symbol('NEVER_SET'),
'name': symbol('NO_VALUE')}
Чтобы получить поведение после обновления, вам нужно истечь category.top_product_id (чтобы он не был включен в INSERT) и fudge category.top_product committed_state (чтобы заставить SQLAlchemy полагать, что значение изменилось и, следовательно, должно вызвать UPDATE).
Во-первых, истекает category.top_product_id, прежде чем сделать category кратковременным:
source_session.expire(category, ["top_product_id"])
Затем выдумка category.top_product committed_state (это может произойти до или после category переходного процесса):
from sqlalchemy import inspect
from sqlalchemy.orm.base import NEVER_SET
inspect(category).committed_state.update(top_product=NEVER_SET)
Полный пример:
from sqlalchemy import Column, ForeignKey, Integer, String, create_engine, inspect
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, make_transient, relationship
from sqlalchemy.orm.base import NEVER_SET
class Base(object):
name = Column(String(50), nullable=False)
Base = declarative_base(cls=Base)
class Category(Base):
__tablename__ = 'categories'
id = Column(Integer, primary_key=True)
top_product_id = Column(Integer, ForeignKey('products.id'))
products = relationship('Product', primaryjoin='Product.category_id == Category.id', back_populates='category', cascade='all', lazy='selectin')
top_product = relationship('Product', primaryjoin='Category.top_product_id == Product.id', post_update=True, cascade='all', lazy='selectin')
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
category_id = Column(Integer, ForeignKey('categories.id'), nullable=False)
category = relationship('Category', primaryjoin='Product.category_id == Category.id', back_populates='products', cascade='all', lazy='selectin')
source_engine = create_engine('sqlite:///')
dest_engine = create_engine('sqlite:///', echo=True)
def fk_pragma_on_connect(dbapi_con, con_record):
dbapi_con.execute('pragma foreign_keys=ON')
from sqlalchemy import event
for engine in [source_engine, dest_engine]:
event.listen(engine, 'connect', fk_pragma_on_connect)
Base.metadata.create_all(bind=source_engine)
Base.metadata.create_all(bind=dest_engine)
source_session = Session(bind=source_engine)
dest_session = Session(bind=dest_engine)
source_category = Category(id=99, name='SomeCategoryName')
source_product = Product(category=source_category, id=100, name='SomeProductName')
source_category.top_product = source_product
source_session.add(source_category)
source_session.commit()
source_session.close()
# If you want to test UPSERTs in dest_session.
# dest_category = Category(id=99, name='PrevCategoryName')
# dest_product = Product(category=dest_category, id=100, name='PrevProductName')
# dest_category.top_product = dest_product
# dest_session.add(dest_category)
# dest_session.commit()
# dest_session.close()
category = source_session.query(Category).filter(Category.id == 99).one()
# Ensure relationship attributes are initialized before we make objects transient.
_ = category.top_product
# source_session.expire(category, ['id']) # only if you want new IDs in dest_session
source_session.expire(category, ['top_product_id'])
for product in category.products:
# Ensure relationship attributes are initialized before we make objects transient.
_ = product.category
# source_session.expire(product, ['id']) # only if you want new IDs in dest_session
# Not strictly needed as long as Product.category is not a post-update relationship.
source_session.expire(product, ['category_id'])
make_transient(category)
inspect(category).committed_state.update(top_product=NEVER_SET)
for product in category.products:
make_transient(product)
# Not strictly needed as long as Product.category is not a post-update relationship.
inspect(product).committed_state.update(category=NEVER_SET)
dest_session.add(category)
# Or, if you want UPSERT (must retain original IDs in this case)
# dest_session.merge(category)
dest_session.flush()
Который производит этот DML в dest_session:
INSERT INTO categories (name, id, top_product_id) VALUES (?, ?, ?)
('SomeCategoryName', 99, None)
INSERT INTO products (name, id, category_id) VALUES (?, ?, ?)
('SomeProductName', 100, 99)
UPDATE categories SET top_product_id=? WHERE categories.id = ?
(100, 99)
Кажется, что make_transient должен сбросить committed_state так, как если бы это был новый объект, но я не думаю.