1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
import sqlalchemy as sa
from pytest import raises
from sqlalchemy_utils import auto_delete_orphans, ImproperlyConfigured
from tests import TestCase
class TestAutoDeleteOrphans(TestCase):
def create_models(self):
tagging = sa.Table(
'tagging',
self.Base.metadata,
sa.Column(
'tag_id',
sa.Integer,
sa.ForeignKey('tag.id', ondelete='cascade'),
primary_key=True
),
sa.Column(
'entry_id',
sa.Integer,
sa.ForeignKey('entry.id', ondelete='cascade'),
primary_key=True
)
)
class Tag(self.Base):
__tablename__ = 'tag'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String(100), unique=True, nullable=False)
def __init__(self, name=None):
self.name = name
class Entry(self.Base):
__tablename__ = 'entry'
id = sa.Column(sa.Integer, primary_key=True)
tags = sa.orm.relationship(
'Tag',
secondary=tagging,
backref='entries'
)
auto_delete_orphans(Entry.tags)
self.Tag = Tag
self.Entry = Entry
def test_orphan_deletion(self):
r1 = self.Entry()
r2 = self.Entry()
r3 = self.Entry()
t1, t2, t3, t4 = (
self.Tag('t1'),
self.Tag('t2'),
self.Tag('t3'),
self.Tag('t4')
)
r1.tags.extend([t1, t2])
r2.tags.extend([t2, t3])
r3.tags.extend([t4])
self.session.add_all([r1, r2, r3])
assert self.session.query(self.Tag).count() == 4
r2.tags.remove(t2)
assert self.session.query(self.Tag).count() == 4
r1.tags.remove(t2)
assert self.session.query(self.Tag).count() == 3
r1.tags.remove(t1)
assert self.session.query(self.Tag).count() == 2
class TestAutoDeleteOrphansWithoutBackref(TestCase):
def create_models(self):
tagging = sa.Table(
'tagging',
self.Base.metadata,
sa.Column(
'tag_id',
sa.Integer,
sa.ForeignKey('tag.id', ondelete='cascade'),
primary_key=True
),
sa.Column(
'entry_id',
sa.Integer,
sa.ForeignKey('entry.id', ondelete='cascade'),
primary_key=True
)
)
class Tag(self.Base):
__tablename__ = 'tag'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String(100), unique=True, nullable=False)
def __init__(self, name=None):
self.name = name
class Entry(self.Base):
__tablename__ = 'entry'
id = sa.Column(sa.Integer, primary_key=True)
tags = sa.orm.relationship(
'Tag',
secondary=tagging
)
self.Entry = Entry
def test_orphan_deletion(self):
with raises(ImproperlyConfigured):
auto_delete_orphans(self.Entry.tags)
|
