-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathPgFunction.py
More file actions
159 lines (112 loc) · 4.2 KB
/
PgFunction.py
File metadata and controls
159 lines (112 loc) · 4.2 KB
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
import hashlib
from ..diff.PgDiffUtils import PgDiffUtils
from ..helpers.OrderedDict import OrderedDict
class PgFunction(object):
def __init__(self):
self.name = ''
self.arguments = []
self.body = None
self.comment = None
def __eq__(self, other):
return self._equals(other, False)
def __ne__(self, other):
return not self._equals(other, False)
def equals(self, other, ignoreFunctionWhitespace):
return self._equals(other, ignoreFunctionWhitespace)
def _equals(self, other, ignoreFunctionWhitespace = False):
if isinstance(other, PgFunction):
if (self.name is None
and other.name is not None or self.name is not None
and self.name != other.name):
return False
if ignoreFunctionWhitespace:
selfBody = ' '.join(self.body.split())
otherBody = ' '.join(other.body.split())
else:
selfBody = self.body
otherBody = other.body
if (selfBody is None
and otherBody is not None or selfBody is not None
and selfBody != otherBody):
return False
if len(self.arguments) != len(other.arguments):
return False
else:
for index, argument in enumerate(self.arguments):
if (argument != other.arguments[index]):
return False
return True
return False
def addArgument(self, argument):
self.arguments.append(argument)
def getCreationSQL(self):
sbSQL = []
sbSQL.append("CREATE OR REPLACE FUNCTION ")
sbSQL.append(PgDiffUtils.getQuotedName(self.name))
sbSQL.append('(')
argumentsString = ', '.join([argument.getDeclaration(True) for argument in self.arguments])
sbSQL.append(argumentsString)
sbSQL.append(") ")
sbSQL.append(self.body)
if self.comment:
sbSQL.append("\n\nCOMMENT ON FUNCTION ")
sbSQL.append(PgDiffUtils.getQuotedName(self.name))
sbSQL.append('(')
sbSQL.append(argumentsString)
sbSQL.append(") IS ")
sbSQL.append(self.comment)
sbSQL.append(';\n')
return ''.join(sbSQL)
def getDropSQL(self):
sbSQL = []
sbSQL.append("DROP FUNCTION ")
sbSQL.append(self.name)
sbSQL.append('(')
addComma = False
for argument in self.arguments:
if "OUT" == argument.mode:
continue
if addComma:
sbSQL.append(", ")
sbSQL.append(argument.getDeclaration(False))
addComma = True
sbSQL.append(");")
return ''.join(sbSQL)
def getSignature(self):
sbSQL = []
sbSQL.append(self.name)
sbSQL.append('(')
addComma = False
for argument in self.arguments:
if argument.mode == "OUT":
continue
if addComma:
sbSQL.append(',')
sbSQL.append(argument.dataType)
addComma = True
sbSQL.append(')')
return hashlib.md5(''.join(sbSQL)).digest().encode("base64")
class Argument(object):
def __init__(self):
self.mode = "IN"
self.name = None
self.dataType = None
self.defaultExpression = None
def __eq__(self, other):
return self.__dict__ == other.__dict__
def __ne__(self, other):
return self.__dict__ != other.__dict__
def getDeclaration(self, includeDefaultValue):
sbSQL = []
if (self.mode is not None and self.mode != "IN"):
sbSQL.append(self.mode)
sbSQL.append(' ')
if self.name is not None and len(self.name) > 0:
sbSQL.append(PgDiffUtils.getQuotedName(self.name))
sbSQL.append(' ')
sbSQL.append(self.dataType)
if (includeDefaultValue and self.defaultExpression is not None
and len(self.defaultExpression) > 0):
sbSQL.append(" = ")
sbSQL.append(self.defaultExpression)
return ''.join(sbSQL)