SQLAlchemy в Python не очищает вторичную (многие-ко-многим) таблицу?
У меня есть отношения многие ко многим междуUser
с иTask
s. Я хочу, чтобы «вторичная таблица» (то есть таблица, которая облегчает отношение «многие ко многим») была очищена при удаленииTask
или жеUser
, Как я могу настроить SQLAlchemy для этого?
Вот пример кода Python, который демонстрирует мою проблему. Примечание. Этот код полностью автономен и требует толькоsqlalchemy
модуль. Если вы скопируете и вставите этот код, вы сможете запустить его без каких-либо побочных эффектов и увидеть то же поведение самостоятельно. Последняя строка скрипта показывает, что соответствующая строка в «вторичной таблице» не была удалена, когда я удалил соответствующую задачу. Все утверждения проходят в этом примере.
from sqlalchemy import create_engine, Column, Integer, Text, Table, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, relationship
Model = declarative_base()
class User(Model):
__tablename__ = 'users'
id = Column('user_id', Integer, primary_key=True)
email = Column('email', Text, unique=True)
def __init__(self, email):
self.email = email
user_tasks = Table('user_tasks', Model.metadata,
Column('user_id', Integer, ForeignKey('users.user_id')),
Column('task_id', Integer, ForeignKey('tasks.task_id')))
class Task(Model):
__tablename__ = 'tasks'
id = Column('task_id', Integer, primary_key=True)
description = Column('description', Text)
assigned_to = relationship('User', secondary=user_tasks, backref='tasks')
def __init__(self, description):
self.description = description
if __name__ == '__main__':
engine = create_engine('sqlite:///:memory:')
Model.metadata.create_all(engine)
s = Session(engine)
the_user = User('user')
s.add(the_user)
s.commit()
assert s.query(User).all() == [the_user]
user_task = Task('user_one task')
user_task.assigned_to.append(the_user)
s.add(user_task)
s.commit()
assert s.query(Task).all() == [user_task]
assert s.query(user_tasks).all() == [(1,1)]
s.query(Task).delete()
s.commit()
assert s.query(Task).all() == []
assert s.query(User).all() == [the_user]
assert s.query(user_tasks).all() == [(1,1)] # I was expecting [] .