[PyGreSQL] Version 3.0 PyGreSQL

Tony Lownds tony at printra.net
Tue Apr 18 11:22:13 EDT 2000


>
>I'm afraid I will need some help here as I don't have the experience with
>DB-SIG API that I need to do this.  All I ask is that yo send any patches
>(context diffs please) to this list so that others can review it before
>I incorporate it.

I've attached the diff. It was only 20 lines shorter than the original.
The changes fix the problems I describe earlier, add a little 
documentation and tweak the names of the exceptions to match the 
spec. I've also attached test suite which tests all the fixes I made.

>
>Yes, see the warning in the new Annonce file.  This is a non-backwards
>compatible change to the API.  I suspect that you are seeing NULLS in
>your database and not blanks.  Try a SELECT testing those columns
>agains NULL and see.
>

You are right, thanks.


>  > Also, there is no Makefile in the -beta.tgz tarball, and the .spec
>>  file has the wrong value in the release macro.
>
>It's kind of hard to include a Makefile when there are so many ways to
>compile it.  Most cases one would use the Python Makefile anyway.  The
>README points the way and is probably the best we can do.  If I included
>a Makefile I would probably get even more complaints when it was done
>wrong for variable definitions of "wrong."

True.... I thought you included the Makefile.pre.in system in version 
2.4 but that's not the case. Its what I use to compile it, fwiw. Very 
civilized. Maybe someone who knows about the new distutils system 
will add support for that, since that seems to be the wave of the 
future.

-Tony
-------------- next part --------------
A non-text attachment was scrubbed...
Name: pgdb.diff
Type: application/mac-binhex40
Size: 12659 bytes
Desc: not available
Url : http://mailman.vex.net/pipermail/pygresql/attachments/20000418/4257ef92/pgdb.bin
-------------- next part --------------
#!/usr/bin/env python
# this test script will exit with a non-zero status if any tests fail,
# and will print some text about the problem. It may also print
# info as it goes along.
#
# started by Tony Lownds 2000/4/17
#

# change this to match your system.
CONNECT = ':postgres'

# some tests will not be fatal, so instead of exiting immediately we set
# the error variable, which is used at the very end of the script
error = 0

# test code begins here.
import traceback
import sys

print "starting the tests."

# step 1. import pgdb (fatal)
try:
   import pgdb
except ImportError:
   try:
     pgdb = 0; del pgdb
     del sys.modules['pgdb']
     import crypt   # this was necessary on my system, until I did -lcrypt
		   # when compiling _pg.so
     import pgdb
     reload(pgdb)
   except ImportError:
     print "FATAL: Cannot import pgdb. Is it on sys.path?"
     print sys.path
     print "Here is the traceback."
     traceback.print_exc()
     sys.exit(1)
   else:
     print """\
Warning: your installation appears to require, as a workaround, import crypt
before an import pgdb. You can fix this by adding -lcrypt when compiling
the _pg extension. """

# step 2. open a connection. (fatal)
try: conn = pgdb.connect(CONNECT)
except:
   print "FATAL: Unable to open a connection. Please check the 
connection string, as coded in %s" % sys.argv[0]
   print "It is %s" % `CONNECT`
   print "Here is the traceback."
   traceback.print_exc()
   sys.exit(1)

# step 3. test the required attributes of the module, and of the connection
# object.
try:
   print "pgdb.apilevel:", pgdb.apilevel
   threadsafety_map = {
   0: 'Threads may not share the module.',
   1: 'Threads may share the module, but not connections.',
   2: 'Threads may share the module and connections.',
   3: 'Threads may share the module, connections and cursors.',
   }
   print "pgdb.threadsafety:", pgdb.threadsafety, 
threadsafety_map[pgdb.threadsafety]
   print "pgdb.paramstyle:", pgdb.paramstyle

   # connection attributes are all methods. Silently test close commit and
   # cursor for existence, and show if rollback exists
   conn.close, conn.commit, conn.cursor
   if hasattr(conn, 'rollback'):
     print "rollback method is supported"
   else:
     print "rollback method is not supported"

except:
   print "ERROR: Something is notright with the module attributes. 
Check out the traceback."
   traceback.print_exc()
   error = 1

# step 4. get a cursor (fatal)
try:
   csr = conn.cursor()
except:
   print "FATAL: Cannot get a cursor, because:"
   traceback.print_exc()
   sys.exit(1)

# step 5. test the cursor's attributes and method existence
# 5.1 initial values of attributes
try: assert csr.description == None
except:
   print "ERROR: cursor.description not initialized properly. (should be None)"
   error = 1
try: assert csr.rowcount == -1
except:
   print "ERROR: cursor.rowcount not initialized properly. (should be -1)"
   error = 1
try: assert csr.arraysize == 1
except:
   print "ERROR: cursor.arraysize not initialized properly. (should be 1)"
   error = 1

# 5.2 test writability of arraysize argument.
try:
   csr.arraysize = 10
   assert csr.arraysize == 10
   csr.arraysize = 11
   assert csr.arraysize == 11
except:
   print "ERROR: cursor.arraysize is not writable"
   traceback.print_exc()

# 5.3 method existence
for meth in ['close', 'execute', 'executemany', 'fetchone', 'fetchmany',
              'fetchall', 'setinputsizes', 'setoutputsize']:
   if not hasattr(csr, meth):
     print "ERROR: Required method %s not present in cursor object" % meth
     error = 1
for meth in ['nextset', 'callproc']:
   if not hasattr(csr, meth):
     print "Optional method %s not present in cursor object" % meth
   else:
     print "Optional method %s is present in cursor object" % meth

# step 6. run a query, and get results.
csr = None    # we'll make a fresh cursor for all tests from now on.
qry = None    # also we'll place the query in here for constistency
# 6.1 run the non-table query and get the results using fetchone()
# also show what execute() returns - spec leaves it open
try:
   csr = conn.cursor()
   qry = "select 'test', 0, NULL, ''"
   ret = csr.execute(qry)
   print "NOTE: cursor.execute(select query) returns: ", `ret`
   results = csr.fetchone()
   # use list() because the spec says
   # 'sequence', so either tuples and lists and other objects are valid.
   assert list(results) == ['test', 0, None, ''], "did not get expected results"
   assert csr.fetchone() == None, "subsequent fetchone call returned 
non-None response"
except:
   print "ERROR: running fetchone on non-table query:", qry
   traceback.print_exc()
   error = 1

# 6.2 run the non-table query and get the results using fetchmany()
try:
   csr = conn.cursor()
   qry = "select 'test', 0, NULL"
   csr.execute(qry)
   results = csr.fetchmany()
   assert map(list, results) == [['test', 0, None]], "did not get 
expected results"
   assert list(csr.fetchmany()) == [], "subsequent call did not return 
empty sequence"
except:
   print "ERROR: running fetchmany on non-table query:", qry
   traceback.print_exc()
   error = 1

# 6.3 run the non-table query and get the results using fetchall()
try:
   csr = conn.cursor()
   qry = "select 'test', 0, NULL"
   csr.execute(qry)
   results = csr.fetchall()
   assert list(results) == [['test', 0, None]]
   # see what happens when we call again - the spec is silent.
   try:
     result = csr.fetchall()
   except:
     print "NOTE: Subsequent call to fetchall() raises an exception:", 
sys.exc_type, sys.exc_value
   else:
     print "NOTE: Subsequent call to fetchall() returns this:", `result`
except:
   print "ERROR: running fetchall on a non-table query:", qry
   traceback.print_exc()
   error = 1

# 6.4 run the non-table query and check the cursor's description and
# rowcount attributes
try:
   csr = conn.cursor()
   qry = "select 'test'"
   csr.execute(qry)
except:
   print """\
Not checking cursor attribute values because query broke, for the same
reasons that have been printed three times above."""
else:
   if not hasattr(csr, 'description'):
     print "ERROR: no description attribute, even after running a query."
     error = 1
   else:
     if len(csr.description) == 1 and len(csr.description[0]) == 7:
       # ok, cursor.description has the right shape
       print "NOTE: cursor.description returns this for a non-table column "\
             ":", csr.description
     else:
       print "ERROR: cursor.description field is not the right shape."
       error = 1


   if not hasattr(csr, 'rowcount'):
     print "ERROR: no rowcount attribute, even after running a query."
     error = 1
   else:
     if csr.rowcount == -1:
       # the spec says a cursor does not have to have a value other than -1
       # until after fetchXXX() - and even then it doesnt have to.
       csr.fetchone()
       if csr.rowcount == -1:
         print "WARNING: cursor.rowcount is not a useful value, even 
for selects"
       elif csr.rowcount == 1:
         print "NOTE: cursor.rowcount is only useful after a call to 
fetchXXX"
       else:
         print "ERROR: expected value of 1 for rowcount:", `csr.rowcount`
         error = 1
     elif csr.rowcount == 1:
       # cursor is behaving well, dont say anything.
       pass
     else:
       print "ERROR: expected value of 1 for rowcount:", `csr.rowcount`
       error = 1

# step 7. make a table for testing purposes (fatal)
try:
   csr = conn.cursor()
   qry = """create table test_table(c1 integer not null, c2 integer, 
c3 varchar)"""
   print "Making table..."
   #csr.execute('begin work')
   csr.execute(qry)
   conn.commit()
except:
   # try to drop then try again
   conn.rollback()
   info = sys.exc_info() # save exception info
   csr = conn.cursor()
   qry2 = "drop table test_table"
   try:
     print "Dropping table first..."
     # I think this is non-standard
     #csr.execute("begin work")
     csr.execute(qry2)
   except:
     print "FATAL: Cannot create or drop the test table. Is it a 
permissions problem?"
     print "query for create:", qry
     print "exception info for create:"
     traceback.print_exception(info[0], info[1], info[2])
     print "query for drop:", qry2
     print "exception info for drop:"
     traceback.print_exc()

     sys.exit(1)
   else:
     conn.commit()   # i believe Postgres (at least ver 6.4.2)
                     # requires the drop be in a seperate transaction
     csr = conn.cursor()
     try:
       print "Creating table again..."
       csr.execute(qry)
       conn.commit()
     except:
       print "FATAL: Cannot create the test table after successfully 
dropping it. Is it a permissions problem?"
       traceback.print_exc()
       sys.exit(1)

# step 8. test queries that return no rows.
try:
   csr = conn.cursor()
   qry = "select * from test_table"
   csr.execute(qry)

   assert list(csr.fetchall()) == [], "fetchall() on a no-rows select 
query should return []"

   print "NOTE: cursor.description", ['is not', 'is'][csr.description == None],\
         "None for query without rows."

except:
   print "ERROR: Some kind of problem with queries that do not return rows."
   print "Check the traceback:"
   traceback.print_exc()
   error = 1

# step 9. insert some data in our test table.
# we try both with column names and without, although any problems related
# to that would not be DB-API related problems...
try:
   csr = conn.cursor()
   csr.execute("insert into test_table(c1,c2,c3) values (1, 1, 'one')")
   csr.execute("insert into test_table values (1, NULL, 'one')")
   csr.execute("insert into test_table values (2, 1, 'one')")
   csr.execute("insert into test_table values (2, NULL, 'two')")
   conn.commit()
except:
   print "ERROR: error inserting rows into test_table:"
   traceback.print_exc()
   error = 1

# step sys.MAXINT: exit.
# please note that we may not get here if a fatal problem has occurred....
if not error:
   print "Thank you, tests are completed successfully."
else:
   print "Sorry, the module has some problems (or the test code does)"
sys.exit(error)


More information about the PyGreSQL mailing list