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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
from itertools import zip_longest
from sqlalchemy import schema
class CompareTable:
def __init__(self, table):
self.table = table
def __eq__(self, other):
if self.table.name != other.name or self.table.schema != other.schema:
return False
for c1, c2 in zip_longest(self.table.c, other.c):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
return False
if CompareColumn(c1) != c2:
return False
return True
# TODO: compare constraints, indexes
def __ne__(self, other):
return not self.__eq__(other)
class CompareColumn:
def __init__(self, column):
self.column = column
def __eq__(self, other):
return (
self.column.name == other.name
and self.column.nullable == other.nullable
)
# TODO: datatypes etc
def __ne__(self, other):
return not self.__eq__(other)
class CompareIndex:
def __init__(self, index, name_only=False):
self.index = index
self.name_only = name_only
def __eq__(self, other):
if self.name_only:
return self.index.name == other.name
else:
return (
str(schema.CreateIndex(self.index))
== str(schema.CreateIndex(other))
and self.index.dialect_kwargs == other.dialect_kwargs
)
def __ne__(self, other):
return not self.__eq__(other)
class CompareCheckConstraint:
def __init__(self, constraint):
self.constraint = constraint
def __eq__(self, other):
return (
isinstance(other, schema.CheckConstraint)
and self.constraint.name == other.name
and (str(self.constraint.sqltext) == str(other.sqltext))
and (other.table.name == self.constraint.table.name)
and other.table.schema == self.constraint.table.schema
)
def __ne__(self, other):
return not self.__eq__(other)
class CompareForeignKey:
def __init__(self, constraint):
self.constraint = constraint
def __eq__(self, other):
r1 = (
isinstance(other, schema.ForeignKeyConstraint)
and self.constraint.name == other.name
and (other.table.name == self.constraint.table.name)
and other.table.schema == self.constraint.table.schema
)
if not r1:
return False
for c1, c2 in zip_longest(self.constraint.columns, other.columns):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
return False
if CompareColumn(c1) != c2:
return False
return True
def __ne__(self, other):
return not self.__eq__(other)
class ComparePrimaryKey:
def __init__(self, constraint):
self.constraint = constraint
def __eq__(self, other):
r1 = (
isinstance(other, schema.PrimaryKeyConstraint)
and self.constraint.name == other.name
and (other.table.name == self.constraint.table.name)
and other.table.schema == self.constraint.table.schema
)
if not r1:
return False
for c1, c2 in zip_longest(self.constraint.columns, other.columns):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
return False
if CompareColumn(c1) != c2:
return False
return True
def __ne__(self, other):
return not self.__eq__(other)
class CompareUniqueConstraint:
def __init__(self, constraint):
self.constraint = constraint
def __eq__(self, other):
r1 = (
isinstance(other, schema.UniqueConstraint)
and self.constraint.name == other.name
and (other.table.name == self.constraint.table.name)
and other.table.schema == self.constraint.table.schema
)
if not r1:
return False
for c1, c2 in zip_longest(self.constraint.columns, other.columns):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
return False
if CompareColumn(c1) != c2:
return False
return True
def __ne__(self, other):
return not self.__eq__(other)
|
