blob: 14f7bb0b18495da580d873c8e48a45211c8e84ef (
about) (
plain)
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
|
# 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
cdef bits_encode(CodecContext settings, WriteBuffer wbuf, obj):
cdef:
Py_buffer pybuf
bint pybuf_used = False
char *buf
ssize_t len
ssize_t bitlen
if cpython.PyBytes_CheckExact(obj):
buf = cpython.PyBytes_AS_STRING(obj)
len = cpython.Py_SIZE(obj)
bitlen = len * 8
elif isinstance(obj, pgproto_types.BitString):
cpython.PyBytes_AsStringAndSize(obj.bytes, &buf, &len)
bitlen = obj.__len__()
else:
cpython.PyObject_GetBuffer(obj, &pybuf, cpython.PyBUF_SIMPLE)
pybuf_used = True
buf = <char*>pybuf.buf
len = pybuf.len
bitlen = len * 8
try:
if bitlen > _MAXINT32:
raise ValueError('bit value too long')
wbuf.write_int32(4 + <int32_t>len)
wbuf.write_int32(<int32_t>bitlen)
wbuf.write_cstr(buf, len)
finally:
if pybuf_used:
cpython.PyBuffer_Release(&pybuf)
cdef bits_decode(CodecContext settings, FRBuffer *buf):
cdef:
int32_t bitlen = hton.unpack_int32(frb_read(buf, 4))
ssize_t buf_len = buf.len
bytes_ = cpython.PyBytes_FromStringAndSize(frb_read_all(buf), buf_len)
return pgproto_types.BitString.frombytes(bytes_, bitlen)
|
