Other differences
-----------------
-When the connection is used as context manager, at the end of the context the
-connection will be closed. In psycopg2 only the transaction is closed, so a
-connection can be used in several contexts, but the behaviour is surprising
-for people used to several other Python classes wrapping resources, such as
-files.
+- When the connection is used as context manager, at the end of the context
+ the connection will be closed. In psycopg2 only the transaction is closed,
+ so a connection can be used in several contexts, but the behaviour is
+ surprising for people used to several other Python classes wrapping
+ resources, such as files.
+
+- `cursor.callproc()` is not implemented. The method has a simplistic
+ semantic which doesn't account for PostgreSQL positional parameters,
+ procedures, set-returning functions. Use a normal
+ `~psycopg3.Cursor.execute()` with :sql:`SELECT function_name(...)` or
+ :sql:`CALL procedure_name(...)` instead.
What's new in psycopg3
import sys
from types import TracebackType
from typing import Any, AsyncIterator, Callable, Generic, Iterator, List
-from typing import Mapping, Optional, Sequence, Type, TYPE_CHECKING, Union
+from typing import Optional, Sequence, Type, TYPE_CHECKING
from operator import attrgetter
from contextlib import contextmanager
from . import errors as e
from . import pq
-from . import sql
from .oids import builtins
from .copy import Copy, AsyncCopy
from .proto import ConnectionType, Query, Params, DumpersMap, LoadersMap, PQGen
"the last operation didn't produce a result"
)
- def _callproc_sql(
- self,
- name: Union[str, sql.Identifier],
- args: Optional[Params] = None,
- kwargs: Optional[Mapping[str, Any]] = None,
- ) -> sql.Composable:
- if args and not isinstance(args, (Sequence, Mapping)):
- raise TypeError(
- f"callproc args should be a sequence or a mapping,"
- f" got {type(args).__name__}"
- )
- if isinstance(args, Mapping) and kwargs:
- raise TypeError(
- "callproc supports only one args sequence and one kwargs mapping"
- )
-
- if not kwargs and isinstance(args, Mapping):
- kwargs = args
- args = None
-
- if kwargs and not isinstance(kwargs, Mapping):
- raise TypeError(
- f"callproc kwargs should be a mapping,"
- f" got {type(kwargs).__name__}"
- )
-
- qparts: List[sql.Composable] = [
- sql.SQL("select * from "),
- name if isinstance(name, sql.Identifier) else sql.Identifier(name),
- sql.SQL("("),
- ]
-
- if args:
- for i, item in enumerate(args):
- if i:
- qparts.append(sql.SQL(","))
- qparts.append(sql.Literal(item))
-
- if kwargs:
- for i, (k, v) in enumerate(kwargs.items()):
- if i:
- qparts.append(sql.SQL(","))
- qparts.extend(
- [sql.Identifier(k), sql.SQL(":="), sql.Literal(v)]
- )
-
- qparts.append(sql.SQL(")"))
- return sql.Composed(qparts)
-
def _check_copy_results(self, results: Sequence["PGresult"]) -> None:
"""
Check that the value returned in a copy() operation is a legit COPY.
return self
- def callproc(
- self,
- name: Union[str, sql.Identifier],
- args: Optional[Params] = None,
- kwargs: Optional[Mapping[str, Any]] = None,
- ) -> Optional[Params]:
- """
- Call a stored procedure to the database.
- """
- self.execute(self._callproc_sql(name, args))
- return args
-
def fetchone(self) -> Optional[Sequence[Any]]:
"""
Return the next record from the current recordset.
return self
- async def callproc(
- self,
- name: Union[str, sql.Identifier],
- args: Optional[Params] = None,
- kwargs: Optional[Mapping[str, Any]] = None,
- ) -> Optional[Params]:
- await self.execute(self._callproc_sql(name, args, kwargs))
- return args
-
async def fetchone(self) -> Optional[Sequence[Any]]:
self._check_result()
rv = self._transformer.load_row(self._pos)
import gc
import pytest
import weakref
-from collections import namedtuple
import psycopg3
-from psycopg3 import sql
from psycopg3.oids import builtins
cur.executemany(query, [(10, "hello"), (20, "world")])
-def test_callproc_args(conn):
- cur = conn.cursor()
- cur.execute(
- """
- create function testfunc(a int, b text) returns text[] language sql as
- 'select array[$1::text, $2]'
- """
- )
- assert cur.callproc("testfunc", [10, "twenty"]) == [10, "twenty"]
- assert cur.fetchone() == (["10", "twenty"],)
-
-
-def test_callproc_badparam(conn):
- cur = conn.cursor()
- with pytest.raises(TypeError):
- cur.callproc("lower", 42)
- with pytest.raises(TypeError):
- cur.callproc(42, ["lower"])
-
-
-def make_testfunc(conn):
- # This parameter name tests for injection and quote escaping
- paramname = """Robert'); drop table "students" --"""
- procname = "randall"
-
- # Set up the temporary function
- stmt = (
- sql.SQL(
- """
- create function {}({} numeric) returns numeric language sql as
- 'select $1 * $1'
- """
- )
- .format(sql.Identifier(procname), sql.Identifier(paramname))
- .as_string(conn)
- .encode(conn.client_encoding)
- )
-
- # execute regardless of sync/async conn
- conn.pgconn.exec_(stmt)
-
- return namedtuple("Thang", "name, param")(procname, paramname)
-
-
-def test_callproc_dict(conn):
-
- testfunc = make_testfunc(conn)
- cur = conn.cursor()
-
- cur.callproc(testfunc.name, [2])
- assert cur.fetchone() == (4,)
- cur.callproc(testfunc.name, {testfunc.param: 2})
- assert cur.fetchone() == (4,)
- cur.callproc(sql.Identifier(testfunc.name), {testfunc.param: 2})
- assert cur.fetchone() == (4,)
-
-
-@pytest.mark.parametrize(
- "args, exc",
- [
- ({"_p": 2, "foo": "bar"}, psycopg3.ProgrammingError),
- ({"_p": "two"}, psycopg3.DataError),
- ({"bj\xc3rn": 2}, psycopg3.ProgrammingError),
- ({3: 2}, TypeError),
- ({(): 2}, TypeError),
- ],
-)
-def test_callproc_dict_bad(conn, args, exc):
- testfunc = make_testfunc(conn)
- if "_p" in args:
- args[testfunc.param] = args.pop("_p")
-
- cur = conn.cursor()
- with pytest.raises(exc):
- cur.callproc(testfunc.name, args)
-
-
def test_rowcount(conn):
cur = conn.cursor()
import weakref
import psycopg3
-from psycopg3 import sql
-
-from .test_cursor import make_testfunc
pytestmark = pytest.mark.asyncio
await cur.executemany(query, [(10, "hello"), (20, "world")])
-async def test_callproc_args(aconn):
- cur = await aconn.cursor()
- await cur.execute(
- """
- create function testfunc(a int, b text) returns text[] language sql as
- 'select array[$1::text, $2]'
- """
- )
- assert (await cur.callproc("testfunc", [10, "twenty"])) == [10, "twenty"]
- assert (await cur.fetchone()) == (["10", "twenty"],)
-
-
-async def test_callproc_badparam(aconn):
- cur = await aconn.cursor()
- with pytest.raises(TypeError):
- await cur.callproc("lower", 42)
- with pytest.raises(TypeError):
- await cur.callproc(42, ["lower"])
-
-
-async def test_callproc_dict(aconn):
- testfunc = make_testfunc(aconn)
-
- cur = await aconn.cursor()
-
- await cur.callproc(testfunc.name, [2])
- assert (await cur.fetchone()) == (4,)
- await cur.callproc(testfunc.name, {testfunc.param: 2})
- assert await (cur.fetchone()) == (4,)
- await cur.callproc(sql.Identifier(testfunc.name), {testfunc.param: 2})
- assert await (cur.fetchone()) == (4,)
-
-
-@pytest.mark.parametrize(
- "args, exc",
- [
- ({"_p": 2, "foo": "bar"}, psycopg3.ProgrammingError),
- ({"_p": "two"}, psycopg3.DataError),
- ({"bj\xc3rn": 2}, psycopg3.ProgrammingError),
- ({3: 2}, TypeError),
- ({(): 2}, TypeError),
- ],
-)
-async def test_callproc_dict_bad(aconn, args, exc):
- testfunc = make_testfunc(aconn)
- if "_p" in args:
- args[testfunc.param] = args.pop("_p")
-
- cur = await aconn.cursor()
- with pytest.raises(exc):
- await cur.callproc(testfunc.name, args)
-
-
async def test_rowcount(aconn):
cur = await aconn.cursor()
projects / thirdparty / psycopg.git / commitdiff
