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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
import six
try:
StandardError = StandardError
except NameError:
StandardError = Exception
class OperationError(Exception):
pass
from psycopg2cffi._impl.libpq import libpq, ffi
class Warning(StandardError):
pass
class Error(StandardError):
pgerror = None
pgcode = None
cursor = None
_pgres = None
@property
def diag(self):
return Diagnostics(self)
def __del__(self):
if self._pgres:
libpq.PQclear(self._pgres)
self._pgres = None
def __reduce__(self):
t = super(Error, self).__reduce__()
if not isinstance(t, tuple):
return t
# note: in c implementation reduce returns a 2-items tuple;
# in python a 3-items tuple. Maybe the c exception doesn't have a dict?
if len(t) != 3:
return t
d = t[2].copy()
d.pop('cursor', None)
d.pop('_pgres', None)
return (t[0], t[1], d)
def __setstate__(self, state):
self.pgerror = state.get('pgerror')
self.pgcode = state.get('pgcode')
class InterfaceError(Error):
pass
class DatabaseError(Error):
pass
class DataError(DatabaseError):
pass
class OperationalError(DatabaseError):
pass
class IntegrityError(DatabaseError):
pass
class InternalError(DatabaseError):
pass
class ProgrammingError(DatabaseError):
pass
class NotSupportedError(DatabaseError):
pass
class QueryCanceledError(OperationalError):
pass
class TransactionRollbackError(OperationalError):
pass
class Diagnostics(object):
def __init__(self, exc):
self._exc = exc
def _get_field(self, field):
from psycopg2cffi._impl.adapters import bytes_to_ascii
if self._exc and self._exc._pgres:
b = libpq.PQresultErrorField(self._exc._pgres, field)
if b:
b = ffi.string(b)
if six.PY3: # py2 tests insist on str here
b = bytes_to_ascii(b)
return b
@property
def severity(self):
return self._get_field(libpq.LIBPQ_DIAG_SEVERITY)
@property
def sqlstate(self):
return self._get_field(libpq.LIBPQ_DIAG_SQLSTATE)
@property
def message_primary(self):
return self._get_field(libpq.LIBPQ_DIAG_MESSAGE_PRIMARY)
@property
def message_detail(self):
return self._get_field(libpq.LIBPQ_DIAG_MESSAGE_DETAIL)
@property
def message_hint(self):
return self._get_field(libpq.LIBPQ_DIAG_MESSAGE_HINT)
@property
def statement_position(self):
return self._get_field(libpq.LIBPQ_DIAG_STATEMENT_POSITION)
@property
def internal_position(self):
return self._get_field(libpq.LIBPQ_DIAG_INTERNAL_POSITION)
@property
def internal_query(self):
return self._get_field(libpq.LIBPQ_DIAG_INTERNAL_QUERY)
@property
def context(self):
return self._get_field(libpq.LIBPQ_DIAG_CONTEXT)
@property
def schema_name(self):
return self._get_field(libpq.LIBPQ_DIAG_SCHEMA_NAME)
@property
def table_name(self):
return self._get_field(libpq.LIBPQ_DIAG_TABLE_NAME)
@property
def column_name(self):
return self._get_field(libpq.LIBPQ_DIAG_COLUMN_NAME)
@property
def datatype_name(self):
return self._get_field(libpq.LIBPQ_DIAG_DATATYPE_NAME)
@property
def constraint_name(self):
return self._get_field(libpq.LIBPQ_DIAG_CONSTRAINT_NAME)
@property
def source_file(self):
return self._get_field(libpq.LIBPQ_DIAG_SOURCE_FILE)
@property
def source_line(self):
return self._get_field(libpq.LIBPQ_DIAG_SOURCE_LINE)
@property
def source_function(self):
return self._get_field(libpq.LIBPQ_DIAG_SOURCE_FUNCTION)
|
