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
|
# Copyright (C) 2016-present the asyncpg authors and contributors
# <see AUTHORS file>
#
# This module is part of asyncpg and is released under
# the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0
from asyncpg import exceptions
@cython.final
cdef class ConnectionSettings(pgproto.CodecContext):
def __cinit__(self, conn_key):
self._encoding = 'utf-8'
self._is_utf8 = True
self._settings = {}
self._codec = codecs.lookup('utf-8')
self._data_codecs = DataCodecConfig(conn_key)
cdef add_setting(self, str name, str val):
self._settings[name] = val
if name == 'client_encoding':
py_enc = get_python_encoding(val)
self._codec = codecs.lookup(py_enc)
self._encoding = self._codec.name
self._is_utf8 = self._encoding == 'utf-8'
cdef is_encoding_utf8(self):
return self._is_utf8
cpdef get_text_codec(self):
return self._codec
cpdef inline register_data_types(self, types):
self._data_codecs.add_types(types)
cpdef inline add_python_codec(self, typeoid, typename, typeschema,
typeinfos, typekind, encoder, decoder,
format):
cdef:
ServerDataFormat _format
ClientExchangeFormat xformat
if format == 'binary':
_format = PG_FORMAT_BINARY
xformat = PG_XFORMAT_OBJECT
elif format == 'text':
_format = PG_FORMAT_TEXT
xformat = PG_XFORMAT_OBJECT
elif format == 'tuple':
_format = PG_FORMAT_ANY
xformat = PG_XFORMAT_TUPLE
else:
raise exceptions.InterfaceError(
'invalid `format` argument, expected {}, got {!r}'.format(
"'text', 'binary' or 'tuple'", format
))
self._data_codecs.add_python_codec(typeoid, typename, typeschema,
typekind, typeinfos,
encoder, decoder,
_format, xformat)
cpdef inline remove_python_codec(self, typeoid, typename, typeschema):
self._data_codecs.remove_python_codec(typeoid, typename, typeschema)
cpdef inline clear_type_cache(self):
self._data_codecs.clear_type_cache()
cpdef inline set_builtin_type_codec(self, typeoid, typename, typeschema,
typekind, alias_to, format):
cdef:
ServerDataFormat _format
if format is None:
_format = PG_FORMAT_ANY
elif format == 'binary':
_format = PG_FORMAT_BINARY
elif format == 'text':
_format = PG_FORMAT_TEXT
else:
raise exceptions.InterfaceError(
'invalid `format` argument, expected {}, got {!r}'.format(
"'text' or 'binary'", format
))
self._data_codecs.set_builtin_type_codec(typeoid, typename, typeschema,
typekind, alias_to, _format)
cpdef inline Codec get_data_codec(self, uint32_t oid,
ServerDataFormat format=PG_FORMAT_ANY,
bint ignore_custom_codec=False):
return self._data_codecs.get_codec(oid, format, ignore_custom_codec)
def __getattr__(self, name):
if not name.startswith('_'):
try:
return self._settings[name]
except KeyError:
raise AttributeError(name) from None
return object.__getattribute__(self, name)
def __repr__(self):
return '<ConnectionSettings {!r}>'.format(self._settings)
|
