def pgresult(self) -> Optional["PGresult"]:
return self._pgresult
- @pgresult.setter
- def pgresult(self, result: Optional["PGresult"]) -> None:
+ def set_pgresult(
+ self, result: Optional["PGresult"], set_loaders: bool = True
+ ) -> None:
self._pgresult = result
- rc = self._row_loaders = []
self._ntuples: int
self._nfields: int
if not result:
self._nfields = self._ntuples = 0
+ if set_loaders:
+ self._row_loaders = []
return
nf = self._nfields = result.nfields
self._ntuples = result.ntuples
- for i in range(nf):
- oid = result.ftype(i)
- fmt = result.fformat(i)
- rc.append(self.get_loader(oid, fmt).load) # type: ignore
+ if set_loaders:
+ rc = self._row_loaders = []
+ for i in range(nf):
+ oid = result.ftype(i)
+ fmt = result.fformat(i)
+ rc.append(self.get_loader(oid, fmt).load) # type: ignore
def set_row_types(
self, types: Sequence[int], formats: Sequence[pq.Format]
) -> None:
- rc: List[LoadFunc] = [None] * len(types) # type: ignore[list-item]
- for i in range(len(rc)):
- rc[i] = self.get_loader(types[i], formats[i]).load
+ rc: List[LoadFunc] = []
+ for i in range(len(types)):
+ rc.append(self.get_loader(types[i], formats[i]).load)
self._row_loaders = rc
def pgresult(self, result: Optional["PGresult"]) -> None:
self._pgresult = result
if result and self._tx:
- self._tx.pgresult = result
+ self._tx.set_pgresult(result)
@property
def description(self) -> Optional[List[Column]]:
def pgresult(self) -> Optional[pq.proto.PGresult]:
...
- @pgresult.setter
- def pgresult(self, result: Optional[pq.proto.PGresult]) -> None:
+ def set_pgresult(
+ self, result: Optional[pq.proto.PGresult], set_loaders: bool = True
+ ) -> None:
...
def set_row_types(
def adapters(self) -> AdaptersMap: ...
@property
def pgresult(self) -> Optional[PGresult]: ...
- @pgresult.setter
- def pgresult(self, result: Optional[PGresult]) -> None: ...
+ def set_pgresult(
+ self, result: Optional["PGresult"], set_loaders: bool = True
+ ) -> None: ...
def set_row_types(
self, types: Sequence[int], formats: Sequence[pq.Format]
) -> None: ...
def pgresult(self) -> Optional[PGresult]:
return self._pgresult
- @pgresult.setter
- def pgresult(self, result: Optional[PGresult]) -> None:
- self.set_pgresult(result)
-
- cdef void set_pgresult(self, pq.PGresult result):
+ cpdef void set_pgresult(self, pq.PGresult result, object set_loaders = True):
self._pgresult = result
if result is None:
self._nfields = self._ntuples = 0
+ if set_loaders:
+ self._row_loaders = []
return
cdef libpq.PGresult *res = self._pgresult.pgresult_ptr
cdef int i
cdef object tmp
- cdef list types = PyList_New(self._nfields)
- cdef list formats = PyList_New(self._nfields)
- for i in range(self._nfields):
- tmp = libpq.PQftype(res, i)
- Py_INCREF(tmp)
- PyList_SET_ITEM(types, i, tmp)
-
- tmp = libpq.PQfformat(res, i)
- Py_INCREF(tmp)
- PyList_SET_ITEM(formats, i, tmp)
-
- self._c_set_row_types(self._nfields, types, formats)
+ cdef list types
+ cdef list formats
+ if set_loaders:
+ types = PyList_New(self._nfields)
+ formats = PyList_New(self._nfields)
+ for i in range(self._nfields):
+ tmp = libpq.PQftype(res, i)
+ Py_INCREF(tmp)
+ PyList_SET_ITEM(types, i, tmp)
+
+ tmp = libpq.PQfformat(res, i)
+ Py_INCREF(tmp)
+ PyList_SET_ITEM(formats, i, tmp)
+
+ self._c_set_row_types(self._nfields, types, formats)
def set_row_types(self,
types: Sequence[int], formats: Sequence[Format]) -> None: