Format,
)
from .encodings import py_codecs
-from .misc import error_message, ConninfoOption, PQerror, PGnotify
+from .misc import ConninfoOption, PQerror, PGnotify, PGresAttDesc
+from .misc import error_message
from . import proto
logger = logging.getLogger(__name__)
"PGconn",
"PGnotify",
"Conninfo",
+ "PGresAttDesc",
"PQerror",
"error_message",
"ConninfoOption",
_fields_: List[Tuple[str, type]] = []
+class PGresAttDesc_struct(Structure):
+ _fields_ = [
+ ("name", c_char_p),
+ ("tableid", Oid),
+ ("columnid", c_int),
+ ("format", c_int),
+ ("typid", Oid),
+ ("typlen", c_int),
+ ("atttypmod", c_int),
+ ]
+
+
PGconn_ptr = POINTER(PGconn_struct)
PGresult_ptr = POINTER(PGresult_struct)
PQconninfoOption_ptr = POINTER(PQconninfoOption_struct)
PGnotify_ptr = POINTER(PGnotify_struct)
PGcancel_ptr = POINTER(PGcancel_struct)
+PGresAttDesc_ptr = POINTER(PGresAttDesc_struct)
# Function definitions as explained in PostgreSQL 12 documentation
PQmakeEmptyPGresult.argtypes = [PGconn_ptr, c_int]
PQmakeEmptyPGresult.restype = PGresult_ptr
+PQsetResultAttrs = pq.PQsetResultAttrs
+PQsetResultAttrs.argtypes = [PGresult_ptr, c_int, PGresAttDesc_ptr]
+PQsetResultAttrs.restype = c_int
+
# 33.12. Notice Processing
relname: bytes
extra: bytes
+class PGresAttDesc_struct:
+ name: bytes
+ tableid: int
+ columnid: int
+ format: int
+ typid: int
+ typlen: int
+ atttypmod: int
+
def PQhostaddr(arg1: Optional[PGconn_struct]) -> bytes: ...
def PQerrorMessage(arg1: Optional[PGconn_struct]) -> bytes: ...
def PQresultErrorMessage(arg1: Optional[PGresult_struct]) -> bytes: ...
def PQnotifies(
arg1: Optional[PGconn_struct],
) -> Optional[pointer[PGnotify_struct]]: ... # type: ignore
+def PQsetResultAttrs(
+ arg1: Optional[PGresult_struct], arg2: int, arg3: Array[PGresAttDesc_struct] # type: ignore
+) -> int: ...
# fmt: off
# autogenerated: start
ctypedef struct PGresult:
pass
- ctypedef struct PGcancel:
- pass
-
- ctypedef struct PGnotify:
- char *relname
- int be_pid
- char *extra
-
ctypedef struct PQconninfoOption:
char *keyword
char *envvar
char *dispchar
int dispsize
+ ctypedef struct PGnotify:
+ char *relname
+ int be_pid
+ char *extra
+
+ ctypedef struct PGcancel:
+ pass
+
+ ctypedef struct PGresAttDesc:
+ char *name
+ Oid tableid
+ int columnid
+ int format
+ Oid typid
+ int typlen
+ int atttypmod
+
# enums
ctypedef enum PostgresPollingStatusType:
void PQfreemem(void *ptr)
void PQconninfoFree(PQconninfoOption *connOptions)
PGresult *PQmakeEmptyPGresult(PGconn *conn, ExecStatusType status)
+ int PQsetResultAttrs(PGresult *res, int numAttributes, PGresAttDesc *attDescs)
int PQlibVersion()
# 33.12. Notice Processing
dispsize: int
+class PGresAttDesc(NamedTuple):
+ name: bytes
+ tableid: int
+ columnid: int
+ format: int
+ typid: int
+ typlen: int
+ atttypmod: int
+
+
def error_message(obj: Union[PGconn, PGresult]) -> str:
"""
Return an error message from a PGconn or PGresult.
DiagnosticField,
Format,
)
-from .misc import error_message, PGnotify, ConninfoOption, PQerror
+from .misc import PGnotify, ConninfoOption, PQerror, PGresAttDesc
+from .misc import error_message
from . import _pq_ctypes as impl
if TYPE_CHECKING:
- from psycopg3 import pq # noqa
+ from . import proto
__impl__ = "ctypes"
def __init__(self, pgconn_ptr: impl.PGconn_struct):
self.pgconn_ptr: Optional[impl.PGconn_struct] = pgconn_ptr
self.notice_handler: Optional[
- Callable[["pq.proto.PGresult"], None]
+ Callable[["proto.PGresult"], None]
] = None
self.notify_handler: Optional[Callable[[PGnotify], None]] = None
def oid_value(self) -> int:
return impl.PQoidValue(self.pgresult_ptr)
+ def set_attributes(self, descriptions: List[PGresAttDesc]) -> None:
+ structs = [
+ impl.PGresAttDesc_struct(*desc) # type: ignore
+ for desc in descriptions
+ ]
+ array = (impl.PGresAttDesc_struct * len(structs))(*structs) # type: ignore
+ rv = impl.PQsetResultAttrs(self.pgresult_ptr, len(structs), array)
+ if rv == 0:
+ raise PQerror("PQsetResultAttrs failed")
+
class PGcancel:
__slots__ = ("pgcancel_ptr",)
from psycopg3.pq.libpq cimport Oid
from psycopg3.errors import OperationalError
-from psycopg3.pq.misc import error_message, PGnotify, ConninfoOption, PQerror
+from psycopg3.pq.misc import PGnotify, ConninfoOption, PQerror, PGresAttDesc
+from psycopg3.pq.misc import error_message
from psycopg3.pq.enums import (
ConnStatus,
PollingStatus,
def oid_value(self) -> int:
return impl.PQoidValue(self.pgresult_ptr)
+ def set_attributes(self, descriptions: List[PGresAttDesc]):
+ cdef int num = len(descriptions)
+ cdef impl.PGresAttDesc *attrs = <impl.PGresAttDesc *>PyMem_Malloc(
+ num * sizeof(impl.PGresAttDesc))
+
+ for i in range(num):
+ descr = descriptions[i]
+ attrs[i].name = descr.name
+ attrs[i].tableid = descr.tableid
+ attrs[i].columnid = descr.columnid
+ attrs[i].format = descr.format
+ attrs[i].typid = descr.typid
+ attrs[i].typlen = descr.typlen
+ attrs[i].atttypmod = descr.atttypmod
+
+ cdef int res = impl.PQsetResultAttrs(self.pgresult_ptr, num, attrs);
+ PyMem_Free(attrs)
+ if (res == 0):
+ raise PQerror("PQsetResultAttrs failed")
+
cdef class PGcancel:
def __cinit__(self):
)
if TYPE_CHECKING:
- from .misc import PGnotify, ConninfoOption # noqa
+ from .misc import PGnotify, ConninfoOption, PGresAttDesc
class PGconn(Protocol):
def oid_value(self) -> int:
...
+ def set_attributes(self, descriptions: List["PGresAttDesc"]) -> None:
+ ...
+
class PGcancel(Protocol):
def free(self) -> None:
assert pq.error_message(res) == "no details available"
pgconn.finish()
assert "NULL" in pq.error_message(pgconn)
+
+
+def test_make_empty_result(pgconn):
+ pgconn.exec_(b"wat")
+ res = pgconn.make_empty_result(pq.ExecStatus.FATAL_ERROR)
+ assert res.status == pq.ExecStatus.FATAL_ERROR
+ assert b"wat" in res.error_message
+
+ pgconn.finish()
+ res = pgconn.make_empty_result(pq.ExecStatus.FATAL_ERROR)
+ assert res.status == pq.ExecStatus.FATAL_ERROR
+ assert res.error_message == b""
+
+
+def test_result_set_attrs(pgconn):
+ res = pgconn.make_empty_result(pq.ExecStatus.COPY_OUT)
+ assert res.status == pq.ExecStatus.COPY_OUT
+
+ attrs = [
+ pq.PGresAttDesc(b"an_int", 0, 0, 0, 23, 0, 0),
+ pq.PGresAttDesc(b"a_num", 0, 0, 0, 1700, 0, 0),
+ pq.PGresAttDesc(b"a_bin_text", 0, 0, 1, 25, 0, 0),
+ ]
+ res.set_attributes(attrs)
+ assert res.nfields == 3
+
+ assert res.fname(0) == b"an_int"
+ assert res.fname(1) == b"a_num"
+ assert res.fname(2) == b"a_bin_text"
+
+ assert res.fformat(0) == 0
+ assert res.fformat(1) == 0
+ assert res.fformat(2) == 1
+
+ assert res.ftype(0) == 23
+ assert res.ftype(1) == 1700
+ assert res.ftype(2) == 25
+
+ with pytest.raises(pq.PQerror):
+ res.set_attributes(attrs)
cancel.free()
-def test_make_empty_result(pgconn):
- pgconn.exec_(b"wat")
- res = pgconn.make_empty_result(pq.ExecStatus.FATAL_ERROR)
- assert res.status == pq.ExecStatus.FATAL_ERROR
- assert b"wat" in res.error_message
-
- pgconn.finish()
- res = pgconn.make_empty_result(pq.ExecStatus.FATAL_ERROR)
- assert res.status == pq.ExecStatus.FATAL_ERROR
- assert res.error_message == b""
-
-
def test_notify(pgconn):
assert pgconn.notifies() is None
projects / thirdparty / psycopg.git / commitdiff
