From 4a52a71956a8d46fcb7294ac71734504bb09bcc2 Mon Sep 17 00:00:00 2001 From: S. Solomon Darnell Date: Fri, 28 Mar 2025 21:52:21 -0500 Subject: two version of R2R are here --- .../site-packages/asyncpg/pgproto/codecs/int.pyx | 144 +++++++++++++++++++++ 1 file changed, 144 insertions(+) create mode 100644 .venv/lib/python3.12/site-packages/asyncpg/pgproto/codecs/int.pyx (limited to '.venv/lib/python3.12/site-packages/asyncpg/pgproto/codecs/int.pyx') diff --git a/.venv/lib/python3.12/site-packages/asyncpg/pgproto/codecs/int.pyx b/.venv/lib/python3.12/site-packages/asyncpg/pgproto/codecs/int.pyx new file mode 100644 index 00000000..99972444 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/asyncpg/pgproto/codecs/int.pyx @@ -0,0 +1,144 @@ +# Copyright (C) 2016-present the asyncpg authors and contributors +# +# +# This module is part of asyncpg and is released under +# the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0 + + +cdef bool_encode(CodecContext settings, WriteBuffer buf, obj): + if not cpython.PyBool_Check(obj): + raise TypeError('a boolean is required (got type {})'.format( + type(obj).__name__)) + + buf.write_int32(1) + buf.write_byte(b'\x01' if obj is True else b'\x00') + + +cdef bool_decode(CodecContext settings, FRBuffer *buf): + return frb_read(buf, 1)[0] is b'\x01' + + +cdef int2_encode(CodecContext settings, WriteBuffer buf, obj): + cdef int overflow = 0 + cdef long val + + try: + if type(obj) is not int and hasattr(type(obj), '__int__'): + # Silence a Python warning about implicit __int__ + # conversion. + obj = int(obj) + val = cpython.PyLong_AsLong(obj) + except OverflowError: + overflow = 1 + + if overflow or val < INT16_MIN or val > INT16_MAX: + raise OverflowError('value out of int16 range') + + buf.write_int32(2) + buf.write_int16(val) + + +cdef int2_decode(CodecContext settings, FRBuffer *buf): + return cpython.PyLong_FromLong(hton.unpack_int16(frb_read(buf, 2))) + + +cdef int4_encode(CodecContext settings, WriteBuffer buf, obj): + cdef int overflow = 0 + cdef long val = 0 + + try: + if type(obj) is not int and hasattr(type(obj), '__int__'): + # Silence a Python warning about implicit __int__ + # conversion. + obj = int(obj) + val = cpython.PyLong_AsLong(obj) + except OverflowError: + overflow = 1 + + # "long" and "long long" have the same size for x86_64, need an extra check + if overflow or (sizeof(val) > 4 and (val < INT32_MIN or val > INT32_MAX)): + raise OverflowError('value out of int32 range') + + buf.write_int32(4) + buf.write_int32(val) + + +cdef int4_decode(CodecContext settings, FRBuffer *buf): + return cpython.PyLong_FromLong(hton.unpack_int32(frb_read(buf, 4))) + + +cdef uint4_encode(CodecContext settings, WriteBuffer buf, obj): + cdef int overflow = 0 + cdef unsigned long val = 0 + + try: + if type(obj) is not int and hasattr(type(obj), '__int__'): + # Silence a Python warning about implicit __int__ + # conversion. + obj = int(obj) + val = cpython.PyLong_AsUnsignedLong(obj) + except OverflowError: + overflow = 1 + + # "long" and "long long" have the same size for x86_64, need an extra check + if overflow or (sizeof(val) > 4 and val > UINT32_MAX): + raise OverflowError('value out of uint32 range') + + buf.write_int32(4) + buf.write_int32(val) + + +cdef uint4_decode(CodecContext settings, FRBuffer *buf): + return cpython.PyLong_FromUnsignedLong( + hton.unpack_int32(frb_read(buf, 4))) + + +cdef int8_encode(CodecContext settings, WriteBuffer buf, obj): + cdef int overflow = 0 + cdef long long val + + try: + if type(obj) is not int and hasattr(type(obj), '__int__'): + # Silence a Python warning about implicit __int__ + # conversion. + obj = int(obj) + val = cpython.PyLong_AsLongLong(obj) + except OverflowError: + overflow = 1 + + # Just in case for systems with "long long" bigger than 8 bytes + if overflow or (sizeof(val) > 8 and (val < INT64_MIN or val > INT64_MAX)): + raise OverflowError('value out of int64 range') + + buf.write_int32(8) + buf.write_int64(val) + + +cdef int8_decode(CodecContext settings, FRBuffer *buf): + return cpython.PyLong_FromLongLong(hton.unpack_int64(frb_read(buf, 8))) + + +cdef uint8_encode(CodecContext settings, WriteBuffer buf, obj): + cdef int overflow = 0 + cdef unsigned long long val = 0 + + try: + if type(obj) is not int and hasattr(type(obj), '__int__'): + # Silence a Python warning about implicit __int__ + # conversion. + obj = int(obj) + val = cpython.PyLong_AsUnsignedLongLong(obj) + except OverflowError: + overflow = 1 + + # Just in case for systems with "long long" bigger than 8 bytes + if overflow or (sizeof(val) > 8 and val > UINT64_MAX): + raise OverflowError('value out of uint64 range') + + buf.write_int32(8) + buf.write_int64(val) + + +cdef uint8_decode(CodecContext settings, FRBuffer *buf): + return cpython.PyLong_FromUnsignedLongLong( + hton.unpack_int64(frb_read(buf, 8))) \ No newline at end of file -- cgit v1.2.3