Changeset 834
- Timestamp:
- 03/12/07 11:13:58 (2 years ago)
- Files:
-
- pg8000/trunk/pg8000-dbapi-test.py (added)
- pg8000/trunk/pg8000.py (modified) (4 diffs)
- pg8000/trunk/pg8000-test.py (modified) (1 diff)
Legend:
- Unmodified
- Added
- Removed
- Modified
- Copied
- Moved
pg8000/trunk/pg8000.py
r833 r834 94 94 apilevel = "2.0" 95 95 threadsafety = 3 96 paramstyle = ' none-of-the-above'96 paramstyle = 'format' # paramstyle can be changed to any DB-API paramstyle 97 97 98 98 def convert_paramstyle(src_style, query, args): … … 274 274 275 275 def execute(self, operation, args=()): 276 self.cursor.execute(operation, *args) 276 new_query, new_args = DBAPI.convert_paramstyle(DBAPI.paramstyle, operation, args) 277 self.cursor.execute(new_query, *new_args) 277 278 278 279 def executemany(self, operation, parameter_sets): … … 309 310 310 311 def cursor(self): 311 return CursorWrapper(self.conn)312 return DBAPI.CursorWrapper(self.conn) 312 313 313 314 def commit(self): … … 332 333 333 334 def connect(user, host=None, unix_sock=None, port=5432, database=None, password=None, socket_timeout=60): 334 return ConnectionWrapper(user=user, host=host, unix_sock=unix_sock, 335 port=port, database=database, password=password, 336 socket_timeout=socket_timeout) 335 return DBAPI.ConnectionWrapper(user=user, host=host, 336 unix_sock=unix_sock, port=port, database=database, 337 password=password, socket_timeout=socket_timeout) 338 connect = staticmethod(connect) 337 339 338 340 pg8000/trunk/pg8000-test.py
r833 r834 6 6 7 7 import pg8000 8 9 print "testing convert_paramstyle"10 11 new_query, new_args = pg8000.DBAPI.convert_paramstyle("qmark", "SELECT ?, ?, \"field_?\" FROM t WHERE a='say ''what?''' AND b=? AND c=E'?\\'test\\'?'", (1, 2, 3))12 assert new_query == "SELECT $1, $2, \"field_?\" FROM t WHERE a='say ''what?''' AND b=$3 AND c=E'?\\'test\\'?'"13 assert new_args == (1, 2, 3)14 15 new_query, new_args = pg8000.DBAPI.convert_paramstyle("qmark", "SELECT ?, ?, * FROM t WHERE a=? AND b='are you ''sure?'", (1, 2, 3))16 assert new_query == "SELECT $1, $2, * FROM t WHERE a=$3 AND b='are you ''sure?'"17 assert new_args == (1, 2, 3)18 19 new_query, new_args = pg8000.DBAPI.convert_paramstyle("numeric", "SELECT :2, :1, * FROM t WHERE a=:3", (1, 2, 3))20 assert new_query == "SELECT $2, $1, * FROM t WHERE a=$3"21 assert new_args == (1, 2, 3)22 23 new_query, new_args = pg8000.DBAPI.convert_paramstyle("named", "SELECT :f2, :f1 FROM t WHERE a=:f2", {"f2": 1, "f1": 2})24 assert new_query == "SELECT $1, $2 FROM t WHERE a=$1"25 assert new_args == (1, 2)26 27 new_query, new_args = pg8000.DBAPI.convert_paramstyle("format", "SELECT %s, %s, \"f1_%%\", E'txt_%%' FROM t WHERE a=%s AND b='75%%'", (1, 2, 3))28 assert new_query == "SELECT $1, $2, \"f1_%\", E'txt_%' FROM t WHERE a=$3 AND b='75%'"29 assert new_args == (1, 2, 3)30 31 new_query, new_args = pg8000.DBAPI.convert_paramstyle("pyformat", "SELECT %(f2)s, %(f1)s, \"f1_%%\", E'txt_%%' FROM t WHERE a=%(f2)s AND b='75%%'", {"f2": 1, "f1": 2, "f3": 3})32 assert new_query == "SELECT $1, $2, \"f1_%\", E'txt_%' FROM t WHERE a=$1 AND b='75%'"33 assert new_args == (1, 2)34 8 35 9 db = pg8000.Connection(host='joy', user='pg8000-test', database='pg8000-test', password='pg8000-test', socket_timeout=5)
