path: root/.venv/lib/python3.12/site-packages/asyncpg/protocol/codecs/base.pyx

# 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 collections.abc import Mapping as MappingABC

import asyncpg
from asyncpg import exceptions


cdef void* binary_codec_map[(MAXSUPPORTEDOID + 1) * 2]
cdef void* text_codec_map[(MAXSUPPORTEDOID + 1) * 2]
cdef dict EXTRA_CODECS = {}


@cython.final
cdef class Codec:

    def __cinit__(self, uint32_t oid):
        self.oid = oid
        self.type = CODEC_UNDEFINED

    cdef init(
        self,
        str name,
        str schema,
        str kind,
        CodecType type,
        ServerDataFormat format,
        ClientExchangeFormat xformat,
        encode_func c_encoder,
        decode_func c_decoder,
        Codec base_codec,
        object py_encoder,
        object py_decoder,
        Codec element_codec,
        tuple element_type_oids,
        object element_names,
        list element_codecs,
        Py_UCS4 element_delimiter,
    ):

        self.name = name
        self.schema = schema
        self.kind = kind
        self.type = type
        self.format = format
        self.xformat = xformat
        self.c_encoder = c_encoder
        self.c_decoder = c_decoder
        self.base_codec = base_codec
        self.py_encoder = py_encoder
        self.py_decoder = py_decoder
        self.element_codec = element_codec
        self.element_type_oids = element_type_oids
        self.element_codecs = element_codecs
        self.element_delimiter = element_delimiter
        self.element_names = element_names

        if base_codec is not None:
            if c_encoder != NULL or c_decoder != NULL:
                raise exceptions.InternalClientError(
                    'base_codec is mutually exclusive with c_encoder/c_decoder'
                )

        if element_names is not None:
            self.record_desc = record.ApgRecordDesc_New(
                element_names, tuple(element_names))
        else:
            self.record_desc = None

        if type == CODEC_C:
            self.encoder = <codec_encode_func>&self.encode_scalar
            self.decoder = <codec_decode_func>&self.decode_scalar
        elif type == CODEC_ARRAY:
            if format == PG_FORMAT_BINARY:
                self.encoder = <codec_encode_func>&self.encode_array
                self.decoder = <codec_decode_func>&self.decode_array
            else:
                self.encoder = <codec_encode_func>&self.encode_array_text
                self.decoder = <codec_decode_func>&self.decode_array_text
        elif type == CODEC_RANGE:
            if format != PG_FORMAT_BINARY:
                raise exceptions.UnsupportedClientFeatureError(
                    'cannot decode type "{}"."{}": text encoding of '
                    'range types is not supported'.format(schema, name))
            self.encoder = <codec_encode_func>&self.encode_range
            self.decoder = <codec_decode_func>&self.decode_range
        elif type == CODEC_MULTIRANGE:
            if format != PG_FORMAT_BINARY:
                raise exceptions.UnsupportedClientFeatureError(
                    'cannot decode type "{}"."{}": text encoding of '
                    'range types is not supported'.format(schema, name))
            self.encoder = <codec_encode_func>&self.encode_multirange
            self.decoder = <codec_decode_func>&self.decode_multirange
        elif type == CODEC_COMPOSITE:
            if format != PG_FORMAT_BINARY:
                raise exceptions.UnsupportedClientFeatureError(
                    'cannot decode type "{}"."{}": text encoding of '
                    'composite types is not supported'.format(schema, name))
            self.encoder = <codec_encode_func>&self.encode_composite
            self.decoder = <codec_decode_func>&self.decode_composite
        elif type == CODEC_PY:
            self.encoder = <codec_encode_func>&self.encode_in_python
            self.decoder = <codec_decode_func>&self.decode_in_python
        else:
            raise exceptions.InternalClientError(
                'unexpected codec type: {}'.format(type))

    cdef Codec copy(self):
        cdef Codec codec

        codec = Codec(self.oid)
        codec.init(self.name, self.schema, self.kind,
                   self.type, self.format, self.xformat,
                   self.c_encoder, self.c_decoder, self.base_codec,
                   self.py_encoder, self.py_decoder,
                   self.element_codec,
                   self.element_type_oids, self.element_names,
                   self.element_codecs, self.element_delimiter)

        return codec

    cdef encode_scalar(self, ConnectionSettings settings, WriteBuffer buf,
                       object obj):
        self.c_encoder(settings, buf, obj)

    cdef encode_array(self, ConnectionSettings settings, WriteBuffer buf,
                      object obj):
        array_encode(settings, buf, obj, self.element_codec.oid,
                     codec_encode_func_ex,
                     <void*>(<cpython.PyObject>self.element_codec))

    cdef encode_array_text(self, ConnectionSettings settings, WriteBuffer buf,
                           object obj):
        return textarray_encode(settings, buf, obj,
                                codec_encode_func_ex,
                                <void*>(<cpython.PyObject>self.element_codec),
                                self.element_delimiter)

    cdef encode_range(self, ConnectionSettings settings, WriteBuffer buf,
                      object obj):
        range_encode(settings, buf, obj, self.element_codec.oid,
                     codec_encode_func_ex,
                     <void*>(<cpython.PyObject>self.element_codec))

    cdef encode_multirange(self, ConnectionSettings settings, WriteBuffer buf,
                           object obj):
        multirange_encode(settings, buf, obj, self.element_codec.oid,
                          codec_encode_func_ex,
                          <void*>(<cpython.PyObject>self.element_codec))

    cdef encode_composite(self, ConnectionSettings settings, WriteBuffer buf,
                          object obj):
        cdef:
            WriteBuffer elem_data
            int i
            list elem_codecs = self.element_codecs
            ssize_t count
            ssize_t composite_size
            tuple rec

        if isinstance(obj, MappingABC):
            # Input is dict-like, form a tuple
            composite_size = len(self.element_type_oids)
            rec = cpython.PyTuple_New(composite_size)

            for i in range(composite_size):
                cpython.Py_INCREF(None)
                cpython.PyTuple_SET_ITEM(rec, i, None)

            for field in obj:
                try:
                    i = self.element_names[field]
                except KeyError:
                    raise ValueError(
                        '{!r} is not a valid element of composite '
                        'type {}'.format(field, self.name)) from None

                item = obj[field]
                cpython.Py_INCREF(item)
                cpython.PyTuple_SET_ITEM(rec, i, item)

            obj = rec

        count = len(obj)
        if count > _MAXINT32:
            raise ValueError('too many elements in composite type record')

        elem_data = WriteBuffer.new()
        i = 0
        for item in obj:
            elem_data.write_int32(<int32_t>self.element_type_oids[i])
            if item is None:
                elem_data.write_int32(-1)
            else:
                (<Codec>elem_codecs[i]).encode(settings, elem_data, item)
            i += 1

        record_encode_frame(settings, buf, elem_data, <int32_t>count)

    cdef encode_in_python(self, ConnectionSettings settings, WriteBuffer buf,
                          object obj):
        data = self.py_encoder(obj)
        if self.xformat == PG_XFORMAT_OBJECT:
            if self.format == PG_FORMAT_BINARY:
                pgproto.bytea_encode(settings, buf, data)
            elif self.format == PG_FORMAT_TEXT:
                pgproto.text_encode(settings, buf, data)
            else:
                raise exceptions.InternalClientError(
                    'unexpected data format: {}'.format(self.format))
        elif self.xformat == PG_XFORMAT_TUPLE:
            if self.base_codec is not None:
                self.base_codec.encode(settings, buf, data)
            else:
                self.c_encoder(settings, buf, data)
        else:
            raise exceptions.InternalClientError(
                'unexpected exchange format: {}'.format(self.xformat))

    cdef encode(self, ConnectionSettings settings, WriteBuffer buf,
                object obj):
        return self.encoder(self, settings, buf, obj)

    cdef decode_scalar(self, ConnectionSettings settings, FRBuffer *buf):
        return self.c_decoder(settings, buf)

    cdef decode_array(self, ConnectionSettings settings, FRBuffer *buf):
        return array_decode(settings, buf, codec_decode_func_ex,
                            <void*>(<cpython.PyObject>self.element_codec))

    cdef decode_array_text(self, ConnectionSettings settings,
                           FRBuffer *buf):
        return textarray_decode(settings, buf, codec_decode_func_ex,
                                <void*>(<cpython.PyObject>self.element_codec),
                                self.element_delimiter)

    cdef decode_range(self, ConnectionSettings settings, FRBuffer *buf):
        return range_decode(settings, buf, codec_decode_func_ex,
                            <void*>(<cpython.PyObject>self.element_codec))

    cdef decode_multirange(self, ConnectionSettings settings, FRBuffer *buf):
        return multirange_decode(settings, buf, codec_decode_func_ex,
                                 <void*>(<cpython.PyObject>self.element_codec))

    cdef decode_composite(self, ConnectionSettings settings,
                          FRBuffer *buf):
        cdef:
            object result
            ssize_t elem_count
            ssize_t i
            int32_t elem_len
            uint32_t elem_typ
            uint32_t received_elem_typ
            Codec elem_codec
            FRBuffer elem_buf

        elem_count = <ssize_t><uint32_t>hton.unpack_int32(frb_read(buf, 4))
        if elem_count != len(self.element_type_oids):
            raise exceptions.OutdatedSchemaCacheError(
                'unexpected number of attributes of composite type: '
                '{}, expected {}'
                    .format(
                        elem_count,
                        len(self.element_type_oids),
                    ),
                schema=self.schema,
                data_type=self.name,
            )
        result = record.ApgRecord_New(asyncpg.Record, self.record_desc, elem_count)
        for i in range(elem_count):
            elem_typ = self.element_type_oids[i]
            received_elem_typ = <uint32_t>hton.unpack_int32(frb_read(buf, 4))

            if received_elem_typ != elem_typ:
                raise exceptions.OutdatedSchemaCacheError(
                    'unexpected data type of composite type attribute {}: '
                    '{!r}, expected {!r}'
                        .format(
                            i,
                            BUILTIN_TYPE_OID_MAP.get(
                                received_elem_typ, received_elem_typ),
                            BUILTIN_TYPE_OID_MAP.get(
                                elem_typ, elem_typ)
                        ),
                    schema=self.schema,
                    data_type=self.name,
                    position=i,
                )

            elem_len = hton.unpack_int32(frb_read(buf, 4))
            if elem_len == -1:
                elem = None
            else:
                elem_codec = self.element_codecs[i]
                elem = elem_codec.decode(
                    settings, frb_slice_from(&elem_buf, buf, elem_len))

            cpython.Py_INCREF(elem)
            record.ApgRecord_SET_ITEM(result, i, elem)

        return result

    cdef decode_in_python(self, ConnectionSettings settings,
                          FRBuffer *buf):
        if self.xformat == PG_XFORMAT_OBJECT:
            if self.format == PG_FORMAT_BINARY:
                data = pgproto.bytea_decode(settings, buf)
            elif self.format == PG_FORMAT_TEXT:
                data = pgproto.text_decode(settings, buf)
            else:
                raise exceptions.InternalClientError(
                    'unexpected data format: {}'.format(self.format))
        elif self.xformat == PG_XFORMAT_TUPLE:
            if self.base_codec is not None:
                data = self.base_codec.decode(settings, buf)
            else:
                data = self.c_decoder(settings, buf)
        else:
            raise exceptions.InternalClientError(
                'unexpected exchange format: {}'.format(self.xformat))

        return self.py_decoder(data)

    cdef inline decode(self, ConnectionSettings settings, FRBuffer *buf):
        return self.decoder(self, settings, buf)

    cdef inline has_encoder(self):
        cdef Codec elem_codec

        if self.c_encoder is not NULL or self.py_encoder is not None:
            return True

        elif (
            self.type == CODEC_ARRAY
            or self.type == CODEC_RANGE
            or self.type == CODEC_MULTIRANGE
        ):
            return self.element_codec.has_encoder()

        elif self.type == CODEC_COMPOSITE:
            for elem_codec in self.element_codecs:
                if not elem_codec.has_encoder():
                    return False
            return True

        else:
            return False

    cdef has_decoder(self):
        cdef Codec elem_codec

        if self.c_decoder is not NULL or self.py_decoder is not None:
            return True

        elif (
            self.type == CODEC_ARRAY
            or self.type == CODEC_RANGE
            or self.type == CODEC_MULTIRANGE
        ):
            return self.element_codec.has_decoder()

        elif self.type == CODEC_COMPOSITE:
            for elem_codec in self.element_codecs:
                if not elem_codec.has_decoder():
                    return False
            return True

        else:
            return False

    cdef is_binary(self):
        return self.format == PG_FORMAT_BINARY

    def __repr__(self):
        return '<Codec oid={} elem_oid={} core={}>'.format(
            self.oid,
            'NA' if self.element_codec is None else self.element_codec.oid,
            has_core_codec(self.oid))

    @staticmethod
    cdef Codec new_array_codec(uint32_t oid,
                               str name,
                               str schema,
                               Codec element_codec,
                               Py_UCS4 element_delimiter):
        cdef Codec codec
        codec = Codec(oid)
        codec.init(name, schema, 'array', CODEC_ARRAY, element_codec.format,
                   PG_XFORMAT_OBJECT, NULL, NULL, None, None, None,
                   element_codec, None, None, None, element_delimiter)
        return codec

    @staticmethod
    cdef Codec new_range_codec(uint32_t oid,
                               str name,
                               str schema,
                               Codec element_codec):
        cdef Codec codec
        codec = Codec(oid)
        codec.init(name, schema, 'range', CODEC_RANGE, element_codec.format,
                   PG_XFORMAT_OBJECT, NULL, NULL, None, None, None,
                   element_codec, None, None, None, 0)
        return codec

    @staticmethod
    cdef Codec new_multirange_codec(uint32_t oid,
                                    str name,
                                    str schema,
                                    Codec element_codec):
        cdef Codec codec
        codec = Codec(oid)
        codec.init(name, schema, 'multirange', CODEC_MULTIRANGE,
                   element_codec.format, PG_XFORMAT_OBJECT, NULL, NULL, None,
                   None, None, element_codec, None, None, None, 0)
        return codec

    @staticmethod
    cdef Codec new_composite_codec(uint32_t oid,
                                   str name,
                                   str schema,
                                   ServerDataFormat format,
                                   list element_codecs,
                                   tuple element_type_oids,
                                   object element_names):
        cdef Codec codec
        codec = Codec(oid)
        codec.init(name, schema, 'composite', CODEC_COMPOSITE,
                   format, PG_XFORMAT_OBJECT, NULL, NULL, None, None, None,
                   None, element_type_oids, element_names, element_codecs, 0)
        return codec

    @staticmethod
    cdef Codec new_python_codec(uint32_t oid,
                                str name,
                                str schema,
                                str kind,
                                object encoder,
                                object decoder,
                                encode_func c_encoder,
                                decode_func c_decoder,
                                Codec base_codec,
                                ServerDataFormat format,
                                ClientExchangeFormat xformat):
        cdef Codec codec
        codec = Codec(oid)
        codec.init(name, schema, kind, CODEC_PY, format, xformat,
                   c_encoder, c_decoder, base_codec, encoder, decoder,
                   None, None, None, None, 0)
        return codec


# Encode callback for arrays
cdef codec_encode_func_ex(ConnectionSettings settings, WriteBuffer buf,
                          object obj, const void *arg):
    return (<Codec>arg).encode(settings, buf, obj)


# Decode callback for arrays
cdef codec_decode_func_ex(ConnectionSettings settings, FRBuffer *buf,
                          const void *arg):
    return (<Codec>arg).decode(settings, buf)


cdef uint32_t pylong_as_oid(val) except? 0xFFFFFFFFl:
    cdef:
        int64_t oid = 0
        bint overflow = False

    try:
        oid = cpython.PyLong_AsLongLong(val)
    except OverflowError:
        overflow = True

    if overflow or (oid < 0 or oid > UINT32_MAX):
        raise OverflowError('OID value too large: {!r}'.format(val))

    return <uint32_t>val


cdef class DataCodecConfig:
    def __init__(self, cache_key):
        # Codec instance cache for derived types:
        # composites, arrays, ranges, domains and their combinations.
        self._derived_type_codecs = {}
        # Codec instances set up by the user for the connection.
        self._custom_type_codecs = {}

    def add_types(self, types):
        cdef:
            Codec elem_codec
            list comp_elem_codecs
            ServerDataFormat format
            ServerDataFormat elem_format
            bint has_text_elements
            Py_UCS4 elem_delim

        for ti in types:
            oid = ti['oid']

            if self.get_codec(oid, PG_FORMAT_ANY) is not None:
                continue

            name = ti['name']
            schema = ti['ns']
            array_element_oid = ti['elemtype']
            range_subtype_oid = ti['range_subtype']
            if ti['attrtypoids']:
                comp_type_attrs = tuple(ti['attrtypoids'])
            else:
                comp_type_attrs = None
            base_type = ti['basetype']

            if array_element_oid:
                # Array type (note, there is no separate 'kind' for arrays)

                # Canonicalize type name to "elemtype[]"
                if name.startswith('_'):
                    name = name[1:]
                name = '{}[]'.format(name)

                elem_codec = self.get_codec(array_element_oid, PG_FORMAT_ANY)
                if elem_codec is None:
                    elem_codec = self.declare_fallback_codec(
                        array_element_oid, ti['elemtype_name'], schema)

                elem_delim = <Py_UCS4>ti['elemdelim'][0]

                self._derived_type_codecs[oid, elem_codec.format] = \
                    Codec.new_array_codec(
                        oid, name, schema, elem_codec, elem_delim)

            elif ti['kind'] == b'c':
                # Composite type

                if not comp_type_attrs:
                    raise exceptions.InternalClientError(
                        f'type record missing field types for composite {oid}')

                comp_elem_codecs = []
                has_text_elements = False

                for typoid in comp_type_attrs:
                    elem_codec = self.get_codec(typoid, PG_FORMAT_ANY)
                    if elem_codec is None:
                        raise exceptions.InternalClientError(
                            f'no codec for composite attribute type {typoid}')
                    if elem_codec.format is PG_FORMAT_TEXT:
                        has_text_elements = True
                    comp_elem_codecs.append(elem_codec)

                element_names = collections.OrderedDict()
                for i, attrname in enumerate(ti['attrnames']):
                    element_names[attrname] = i

                # If at least one element is text-encoded, we must
                # encode the whole composite as text.
                if has_text_elements:
                    elem_format = PG_FORMAT_TEXT
                else:
                    elem_format = PG_FORMAT_BINARY

                self._derived_type_codecs[oid, elem_format] = \
                    Codec.new_composite_codec(
                        oid, name, schema, elem_format, comp_elem_codecs,
                        comp_type_attrs, element_names)

            elif ti['kind'] == b'd':
                # Domain type

                if not base_type:
                    raise exceptions.InternalClientError(
                        f'type record missing base type for domain {oid}')

                elem_codec = self.get_codec(base_type, PG_FORMAT_ANY)
                if elem_codec is None:
                    elem_codec = self.declare_fallback_codec(
                        base_type, ti['basetype_name'], schema)

                self._derived_type_codecs[oid, elem_codec.format] = elem_codec

            elif ti['kind'] == b'r':
                # Range type

                if not range_subtype_oid:
                    raise exceptions.InternalClientError(
                        f'type record missing base type for range {oid}')

                elem_codec = self.get_codec(range_subtype_oid, PG_FORMAT_ANY)
                if elem_codec is None:
                    elem_codec = self.declare_fallback_codec(
                        range_subtype_oid, ti['range_subtype_name'], schema)

                self._derived_type_codecs[oid, elem_codec.format] = \
                    Codec.new_range_codec(oid, name, schema, elem_codec)

            elif ti['kind'] == b'm':
                # Multirange type

                if not range_subtype_oid:
                    raise exceptions.InternalClientError(
                        f'type record missing base type for multirange {oid}')

                elem_codec = self.get_codec(range_subtype_oid, PG_FORMAT_ANY)
                if elem_codec is None:
                    elem_codec = self.declare_fallback_codec(
                        range_subtype_oid, ti['range_subtype_name'], schema)

                self._derived_type_codecs[oid, elem_codec.format] = \
                    Codec.new_multirange_codec(oid, name, schema, elem_codec)

            elif ti['kind'] == b'e':
                # Enum types are essentially text
                self._set_builtin_type_codec(oid, name, schema, 'scalar',
                                             TEXTOID, PG_FORMAT_ANY)
            else:
                self.declare_fallback_codec(oid, name, schema)

    def add_python_codec(self, typeoid, typename, typeschema, typekind,
                         typeinfos, encoder, decoder, format, xformat):
        cdef:
            Codec core_codec = None
            encode_func c_encoder = NULL
            decode_func c_decoder = NULL
            Codec base_codec = None
            uint32_t oid = pylong_as_oid(typeoid)
            bint codec_set = False

        # Clear all previous overrides (this also clears type cache).
        self.remove_python_codec(typeoid, typename, typeschema)

        if typeinfos:
            self.add_types(typeinfos)

        if format == PG_FORMAT_ANY:
            formats = (PG_FORMAT_TEXT, PG_FORMAT_BINARY)
        else:
            formats = (format,)

        for fmt in formats:
            if xformat == PG_XFORMAT_TUPLE:
                if typekind == "scalar":
                    core_codec = get_core_codec(oid, fmt, xformat)
                    if core_codec is None:
                        continue
                    c_encoder = core_codec.c_encoder
                    c_decoder = core_codec.c_decoder
                elif typekind == "composite":
                    base_codec = self.get_codec(oid, fmt)
                    if base_codec is None:
                        continue

            self._custom_type_codecs[typeoid, fmt] = \
                Codec.new_python_codec(oid, typename, typeschema, typekind,
                                       encoder, decoder, c_encoder, c_decoder,
                                       base_codec, fmt, xformat)
            codec_set = True

        if not codec_set:
            raise exceptions.InterfaceError(
                "{} type does not support the 'tuple' exchange format".format(
                    typename))

    def remove_python_codec(self, typeoid, typename, typeschema):
        for fmt in (PG_FORMAT_BINARY, PG_FORMAT_TEXT):
            self._custom_type_codecs.pop((typeoid, fmt), None)
        self.clear_type_cache()

    def _set_builtin_type_codec(self, typeoid, typename, typeschema, typekind,
                                alias_to, format=PG_FORMAT_ANY):
        cdef:
            Codec codec
            Codec target_codec
            uint32_t oid = pylong_as_oid(typeoid)
            uint32_t alias_oid = 0
            bint codec_set = False

        if format == PG_FORMAT_ANY:
            formats = (PG_FORMAT_BINARY, PG_FORMAT_TEXT)
        else:
            formats = (format,)

        if isinstance(alias_to, int):
            alias_oid = pylong_as_oid(alias_to)
        else:
            alias_oid = BUILTIN_TYPE_NAME_MAP.get(alias_to, 0)

        for format in formats:
            if alias_oid != 0:
                target_codec = self.get_codec(alias_oid, format)
            else:
                target_codec = get_extra_codec(alias_to, format)

            if target_codec is None:
                continue

            codec = target_codec.copy()
            codec.oid = typeoid
            codec.name = typename
            codec.schema = typeschema
            codec.kind = typekind

            self._custom_type_codecs[typeoid, format] = codec
            codec_set = True

        if not codec_set:
            if format == PG_FORMAT_BINARY:
                codec_str = 'binary'
            elif format == PG_FORMAT_TEXT:
                codec_str = 'text'
            else:
                codec_str = 'text or binary'

            raise exceptions.InterfaceError(
                f'cannot alias {typename} to {alias_to}: '
                f'there is no {codec_str} codec for {alias_to}')

    def set_builtin_type_codec(self, typeoid, typename, typeschema, typekind,
                               alias_to, format=PG_FORMAT_ANY):
        self._set_builtin_type_codec(typeoid, typename, typeschema, typekind,
                                     alias_to, format)
        self.clear_type_cache()

    def clear_type_cache(self):
        self._derived_type_codecs.clear()

    def declare_fallback_codec(self, uint32_t oid, str name, str schema):
        cdef Codec codec

        if oid <= MAXBUILTINOID:
            # This is a BKI type, for which asyncpg has no
            # defined codec.  This should only happen for newly
            # added builtin types, for which this version of
            # asyncpg is lacking support.
            #
            raise exceptions.UnsupportedClientFeatureError(
                f'unhandled standard data type {name!r} (OID {oid})')
        else:
            # This is a non-BKI type, and as such, has no
            # stable OID, so no possibility of a builtin codec.
            # In this case, fallback to text format.  Applications
            # can avoid this by specifying a codec for this type
            # using Connection.set_type_codec().
            #
            self._set_builtin_type_codec(oid, name, schema, 'scalar',
                                         TEXTOID, PG_FORMAT_TEXT)

            codec = self.get_codec(oid, PG_FORMAT_TEXT)

        return codec

    cdef inline Codec get_codec(self, uint32_t oid, ServerDataFormat format,
                                bint ignore_custom_codec=False):
        cdef Codec codec

        if format == PG_FORMAT_ANY:
            codec = self.get_codec(
                oid, PG_FORMAT_BINARY, ignore_custom_codec)
            if codec is None:
                codec = self.get_codec(
                    oid, PG_FORMAT_TEXT, ignore_custom_codec)
            return codec
        else:
            if not ignore_custom_codec:
                codec = self.get_custom_codec(oid, PG_FORMAT_ANY)
                if codec is not None:
                    if codec.format != format:
                        # The codec for this OID has been overridden by
                        # set_{builtin}_type_codec with a different format.
                        # We must respect that and not return a core codec.
                        return None
                    else:
                        return codec

            codec = get_core_codec(oid, format)
            if codec is not None:
                return codec
            else:
                try:
                    return self._derived_type_codecs[oid, format]
                except KeyError:
                    return None

    cdef inline Codec get_custom_codec(
        self,
        uint32_t oid,
        ServerDataFormat format
    ):
        cdef Codec codec

        if format == PG_FORMAT_ANY:
            codec = self.get_custom_codec(oid, PG_FORMAT_BINARY)
            if codec is None:
                codec = self.get_custom_codec(oid, PG_FORMAT_TEXT)
        else:
            codec = self._custom_type_codecs.get((oid, format))

        return codec


cdef inline Codec get_core_codec(
        uint32_t oid, ServerDataFormat format,
        ClientExchangeFormat xformat=PG_XFORMAT_OBJECT):
    cdef:
        void *ptr = NULL

    if oid > MAXSUPPORTEDOID:
        return None
    if format == PG_FORMAT_BINARY:
        ptr = binary_codec_map[oid * xformat]
    elif format == PG_FORMAT_TEXT:
        ptr = text_codec_map[oid * xformat]

    if ptr is NULL:
        return None
    else:
        return <Codec>ptr


cdef inline Codec get_any_core_codec(
        uint32_t oid, ServerDataFormat format,
        ClientExchangeFormat xformat=PG_XFORMAT_OBJECT):
    """A version of get_core_codec that accepts PG_FORMAT_ANY."""
    cdef:
        Codec codec

    if format == PG_FORMAT_ANY:
        codec = get_core_codec(oid, PG_FORMAT_BINARY, xformat)
        if codec is None:
            codec = get_core_codec(oid, PG_FORMAT_TEXT, xformat)
    else:
        codec = get_core_codec(oid, format, xformat)

    return codec


cdef inline int has_core_codec(uint32_t oid):
    return binary_codec_map[oid] != NULL or text_codec_map[oid] != NULL


cdef register_core_codec(uint32_t oid,
                         encode_func encode,
                         decode_func decode,
                         ServerDataFormat format,
                         ClientExchangeFormat xformat=PG_XFORMAT_OBJECT):

    if oid > MAXSUPPORTEDOID:
        raise exceptions.InternalClientError(
            'cannot register core codec for OID {}: it is greater '
            'than MAXSUPPORTEDOID ({})'.format(oid, MAXSUPPORTEDOID))

    cdef:
        Codec codec
        str name
        str kind

    name = BUILTIN_TYPE_OID_MAP[oid]
    kind = 'array' if oid in ARRAY_TYPES else 'scalar'

    codec = Codec(oid)
    codec.init(name, 'pg_catalog', kind, CODEC_C, format, xformat,
               encode, decode, None, None, None, None, None, None, None, 0)
    cpython.Py_INCREF(codec)  # immortalize

    if format == PG_FORMAT_BINARY:
        binary_codec_map[oid * xformat] = <void*>codec
    elif format == PG_FORMAT_TEXT:
        text_codec_map[oid * xformat] = <void*>codec
    else:
        raise exceptions.InternalClientError(
            'invalid data format: {}'.format(format))


cdef register_extra_codec(str name,
                          encode_func encode,
                          decode_func decode,
                          ServerDataFormat format):
    cdef:
        Codec codec
        str kind

    kind = 'scalar'

    codec = Codec(INVALIDOID)
    codec.init(name, None, kind, CODEC_C, format, PG_XFORMAT_OBJECT,
               encode, decode, None, None, None, None, None, None, None, 0)
    EXTRA_CODECS[name, format] = codec


cdef inline Codec get_extra_codec(str name, ServerDataFormat format):
    return EXTRA_CODECS.get((name, format))