MyTetra Share
Делитесь знаниями!
Каскадное удаление в SQLAlchemy
18.01.2018
11:50
Автор: br0ke
Текстовые метки: python, sqlalchemy, sql, db, cascade, ondelete, ondelete=cascade
Раздел: Информационные технологии - Python - Библиотеки - SQLAlchemy

import unittest


from sqlalchemy import create_engine

from sqlalchemy import Column, Integer

from sqlalchemy import String

from sqlalchemy import Text

from sqlalchemy import ForeignKey

from sqlalchemy.orm import sessionmaker

from sqlalchemy.orm import scoped_session

from sqlalchemy.orm import relationship

from sqlalchemy.orm import backref

from sqlalchemy.ext.declarative import declarative_base



class PostgresOnDeleteCascadeTest(unittest.TestCase):


def setUp(self):

Session = scoped_session(sessionmaker(expire_on_commit=False))

class Model(object):

query = Session.query_property()


def save(self, session=Session):

session.add(self)

session.commit()

return self


def delete(self, session=Session):

session.delete(self)

session.commit()


Base = declarative_base(cls=Model)


class User(Base):

__tablename__ = 'users'


id = Column(Integer, primary_key=True)

username = Column(String(255), unique=True, nullable=False)


engine = create_engine('postgresql://test@localhost/test', echo=True)

Session.configure(bind=engine)

Base.metadata.bind = engine

User.__table__.create()

self.engine = engine

self.Base = Base

self.Session = Session

self.User = User


def tearDown(self):

Session = self.Session

Base = self.Base

Session.expunge_all()

Session.rollback()

Session.remove()

Base.metadata.drop_all()


# Passes

def test_cascade_delete_using_sql(self):

Base = self.Base

User = self.User

engine = self.engine

class Post(Base):

__tablename__ = 'posts'


id = Column(Integer, primary_key=True)

title = Column(String(255), nullable=False)

body = Column(Text, nullable=False)

user_id = Column(Integer, ForeignKey(User.id, ondelete='CASCADE'),

nullable=False)

user = relationship(User, backref='posts')


Post.__table__.create()

user = User(username='dude').save()

post = Post(user=user, title='foo', body='bar').save()

engine.execute('DELETE FROM users WHERE id = %d' % user.id)

self.assertEqual(User.query.count(), 0)

self.assertEqual(Post.query.count(), 0)


# Error because it tries to set post.user_id = None

def test_cascade_delete_posts_null_constraint(self):

Base = self.Base

User = self.User

class Post(Base):

__tablename__ = 'posts'


id = Column(Integer, primary_key=True)

title = Column(String(255), nullable=False)

body = Column(Text, nullable=False)

user_id = Column(Integer, ForeignKey(User.id, ondelete='CASCADE'),

nullable=False)

user = relationship(User, backref='posts')


Post.__table__.create()

user = User(username='dude').save()

post = Post(user=user, title='foo', body='bar').save()

user.delete()

self.assertEqual(User.query.count(), 0)

self.assertEqual(Post.query.count(), 0)


# Fails to delete post

def test_cascade_delete_posts(self):

Base = self.Base

User = self.User

class Post(Base):

__tablename__ = 'posts'


id = Column(Integer, primary_key=True)

title = Column(String(255), nullable=False)

body = Column(Text, nullable=False)

user_id = Column(Integer, ForeignKey(User.id, ondelete='CASCADE'))

user = relationship(User, backref='posts')


Post.__table__.create()

user = User(username='dude').save()

post = Post(user=user, title='foo', body='bar').save()

user.delete()

self.assertEqual(User.query.count(), 0)

self.assertEqual(Post.query.count(), 0)


# Passes

# The only right way. You need both ForeignKey and backref.

def test_backref_cascade_delete(self):

Base = self.Base

User = self.User

class Post(Base):

__tablename__ = 'posts'


id = Column(Integer, primary_key=True)

title = Column(String(255), nullable=False)

body = Column(Text, nullable=False)

user_id = Column(Integer, ForeignKey(User.id, ondelete='CASCADE'),

nullable=False)

user = relationship(User,

backref=backref('posts', cascade='delete'))


Post.__table__.create()

user = User(username='dude').save()

post = Post(user=user, title='foo', body='bar').save()

user.delete()

self.assertEqual(User.query.count(), 0)

self.assertEqual(Post.query.count(), 0)

 
MyTetra Share v.0.52
Яндекс индекс цитирования