Modified: trunk/pg.py (953 => 954)
--- trunk/pg.py 2019-01-03 20:38:57 UTC (rev 953)
+++ trunk/pg.py 2019-01-03 22:35:30 UTC (rev 954)
@@ -1869,14 +1869,18 @@
return self.query(*self.adapter.format_query(
command, parameters, types, inline))
- def query_prepared(self, name, *args):
+ def query_prepared(self, name=None, *args):
"""Execute a prepared SQL statement.
This works like the query() method, but you need to pass the name of
a prepared statement that you have already created with prepare().
+ If you pass no parameters or pass an empty name, then the last unnamed
+ statement will be executed.
"""
if not self.db:
raise _int_error('Connection is not valid')
+ if name is None:
+ name = ''
if args:
self._do_debug('EXECUTE', name, args)
return self.db.query_prepared(name, args)
@@ -1883,15 +1887,15 @@
self._do_debug('EXECUTE', name)
return self.db.query_prepared(name)
- def prepare(self, name, command):
+ def prepare(self, command, name=None):
"""Create a prepared SQL statement.
- This creates a prepared statement with the given name for the given
- command for later execution with the query_prepared() method.
- The name can be "" to create an unnamed statement, in which case any
- pre-existing unnamed statement is automatically replaced; otherwise
- it is an error if the statement name is already defined in the current
- database session.
+ This creates a prepared statement for the given command with the
+ the given name for later execution with the query_prepared() method.
+ The name can be empty or left out to create an unnamed statement,
+ in which case any pre-existing unnamed statement is automatically
+ replaced; otherwise it is an error if the statement name is already
+ defined in the current database session.
If any parameters are used, they can be referred to in the query as
numbered parameters of the form $1.
@@ -1898,15 +1902,20 @@
"""
if not self.db:
raise _int_error('Connection is not valid')
+ if name is None:
+ name = ''
self._do_debug('prepare', name, command)
return self.db.prepare(name, command)
- def describe_prepared(self, name):
+ def describe_prepared(self, name=None):
"""Describe a prepared SQL statement.
This method returns a Query object describing the result columns of
- the prepared statement with the given name.
+ the prepared statement with the given name. If you do not specify a
+ name, then the last unnamed statement will be described.
"""
+ if name is None:
+ name = ''
return self.db.describe_prepared(name)
def delete_prepared(self, name=None):
@@ -1913,8 +1922,9 @@
"""Delete a prepared SQL statement
This deallocates a previously prepared SQL statement with the given
- name, or deallocates all prepared statements. Prepared statements are
- also deallocated automatically when the current session ends.
+ name, or deallocates all prepared statements if you do not specify a
+ name. Note that prepared statements are also deallocated automatically
+ when the current session ends.
"""
q = "DEALLOCATE %s" % (name or 'ALL',)
self._do_debug(q)
Modified: trunk/tests/test_classic_dbwrapper.py (953 => 954)
--- trunk/tests/test_classic_dbwrapper.py 2019-01-03 20:38:57 UTC (rev 953)
+++ trunk/tests/test_classic_dbwrapper.py 2019-01-03 22:35:30 UTC (rev 954)
@@ -969,10 +969,25 @@
r = f(q, {}).getresult()[0][0]
self.assertEqual(r, 42)
+ def testPrepare(self):
+ p = self.db.prepare
+ self.assertIsNone(p("select 'hello'", 'my query'))
+ self.assertIsNone(p("select 'world'", 'my other query'))
+ self.assertRaises(pg.ProgrammingError,
+ p, 'my query', "select 'hello, too'")
+
+ def testPrepareUnnamed(self):
+ p = self.db.prepare
+ self.assertIsNone(p("select null"))
+ self.assertIsNone(p("select null", None))
+ self.assertIsNone(p("select null", ''))
+ self.assertIsNone(p("select null", name=None))
+ self.assertIsNone(p("select null", name=''))
+
def testQueryPreparedWithoutParams(self):
p = self.db.prepare
- p('q1', "select 17")
- p('q2', "select 42")
+ p("select 17", 'q1')
+ p("select 42", 'q2')
f = self.db.query_prepared
r = f('q1').getresult()[0][0]
self.assertEqual(r, 17)
@@ -981,8 +996,8 @@
def testQueryPreparedWithParams(self):
p = self.db.prepare
- p('sum', "select 1 + $1 + $2 + $3")
- p('cat', "select initcap($1) || ', ' || $2 || '!'")
+ p("select 1 + $1 + $2 + $3", 'sum')
+ p("select initcap($1) || ', ' || $2 || '!'", 'cat')
f = self.db.query_prepared
r = f('sum', 2, 3, 5).getresult()[0][0]
self.assertEqual(r, 11)
@@ -989,20 +1004,42 @@
r = f('cat', 'hello', 'world').getresult()[0][0]
self.assertEqual(r, 'Hello, world!')
- def testPrepare(self):
+ def testQueryPreparedUnnamedWithOutParams(self):
p = self.db.prepare
- self.assertIsNone(p('', "select null"))
- self.assertIsNone(p('myquery', "select 'hello'"))
- self.assertIsNone(p('myquery2', "select 'world'"))
- self.assertRaises(pg.ProgrammingError,
- p, 'myquery', "select 'hello, too'")
+ p("select 'no name'")
+ f = self.db.query_prepared
+ r = f().getresult()[0][0]
+ self.assertEqual(r, 'no name')
+ r = f(None).getresult()[0][0]
+ self.assertEqual(r, 'no name')
+ r = f('').getresult()[0][0]
+ self.assertEqual(r, 'no name')
+ def testQueryPreparedUnnamedWithParams(self):
+ p = self.db.prepare
+ p("select 1 + $1 + $2")
+ f = self.db.query_prepared
+ r = f(None, 2, 3).getresult()[0][0]
+ self.assertEqual(r, 6)
+ r = f('', 2, 3).getresult()[0][0]
+ self.assertEqual(r, 6)
+
def testDescribePrepared(self):
- self.db.prepare('count', 'select 1 as first, 2 as second')
+ self.db.prepare("select 1 as first, 2 as second", 'count')
f = self.db.describe_prepared
r = f('count').listfields()
self.assertEqual(r, ('first', 'second'))
+ def testDescribePreparedUnnamed(self):
+ self.db.prepare("select null as anon")
+ f = self.db.describe_prepared
+ r = f().listfields()
+ self.assertEqual(r, ('anon',))
+ r = f(None).listfields()
+ self.assertEqual(r, ('anon',))
+ r = f('').listfields()
+ self.assertEqual(r, ('anon',))
+
def testDeletePrepared(self):
f = self.db.delete_prepared
f()
@@ -1009,14 +1046,14 @@
e = pg.OperationalError
self.assertRaises(e, f, 'myquery')
p = self.db.prepare
- p('q1', "select 1")
- p('q2', "select 2")
+ p("select 1", 'q1')
+ p("select 2", 'q2')
f('q1')
f('q2')
self.assertRaises(e, f, 'q1')
self.assertRaises(e, f, 'q2')
- p('q1', "select 1")
- p('q2', "select 2")
+ p("select 1", 'q1')
+ p("select 2", 'q2')
f()
self.assertRaises(e, f, 'q1')
self.assertRaises(e, f, 'q2')
