Changeset 333


Ignore:
Timestamp:
Oct 9, 2008, 4:45:29 PM (11 years ago)
Author:
cito
Message:

Allow the DB wrapper to be used with an existing _pg, DB or pgdb connection.

Location:
trunk/module
Files:
2 edited

Legend:

Unmodified
Added
Removed
  • trunk/module/pg.py

    r332 r333  
    66# Improved by Christoph Zwerschke
    77#
    8 # $Id: pg.py,v 1.56 2008-10-09 16:28:23 cito Exp $
     8# $Id: pg.py,v 1.57 2008-10-09 20:45:29 cito Exp $
    99#
    1010
     
    119119
    120120        def __init__(self, *args, **kw):
    121                 self.db = connect(*args, **kw)
    122                 self.dbname = self.db.db
     121                """Create a new connection.
     122
     123                You can pass either the connection parameters or an existing
     124                _pg or pgdb connection. This allows you to use the methods
     125                of the classic pg interface with a DB-API 2 pgdb connection.
     126
     127                """
     128                if not args and len(kw) == 1:
     129                        db = kw.get('db')
     130                elif not kw and len(args) == 1:
     131                        db = args[0]
     132                else:
     133                        db = None
     134                if db:
     135                        if isinstance(db, DB):
     136                                db = db.db
     137                        else:
     138                                try:
     139                                        db = db._cnx
     140                                except AttributeError:
     141                                        pass
     142                if not db or not hasattr(db, 'db') or not hasattr(db, 'query'):
     143                        db = connect(*args, **kw)
     144                        self._closeable = 1
     145                else:
     146                        self._closeable = 0
     147                self.db = db
     148                self.dbname = db.db
    123149                self._attnames = {}
    124150                self._pkeys = {}
     
    153179                """Close the database connection."""
    154180                # Wraps shared library function so we can track state.
    155                 if self.db:
    156                         self.db.close()
    157                         self.db = None
    158                 else:
    159                         raise InternalError('Connection already closed')
     181                if self._closeable:
     182                        if self.db:
     183                                self.db.close()
     184                                self.db = None
     185                        else:
     186                                raise InternalError('Connection already closed')
    160187
    161188        def reopen(self):
     
    166193
    167194                """
    168                 if self.db:
    169                         self.db.close()
    170                 try:
    171                         self.db = connect(*self._args[0], **self._args[1])
    172                 except:
    173                         self.db = None
    174                         raise
     195                # There is no such shared library function.
     196                if self._closeable:
     197                        if self.db:
     198                                self.db.close()
     199                        try:
     200                                self.db = connect(*self._args[0], **self._args[1])
     201                        except:
     202                                self.db = None
     203                                raise
    175204
    176205        def query(self, qstr):
  • trunk/module/test_pg.py

    r328 r333  
    55# Written by Christoph Zwerschke
    66#
    7 # $Id: test_pg.py,v 1.12 2008-09-16 22:29:48 cito Exp $
     7# $Id: test_pg.py,v 1.13 2008-10-09 20:45:29 cito Exp $
    88#
    99
     
    776776                self.db = pg.DB(self.dbname)
    777777
     778        def testExistingConnection(self):
     779                db = pg.DB(self.db.db)
     780                self.assertEqual(self.db.db, db.db)
     781                self.assert_(db.db)
     782                db.close()
     783                self.assert_(db.db)
     784                db.reopen()
     785                self.assert_(db.db)
     786                db.close()
     787                self.assert_(db.db)
     788                db = pg.DB(self.db)
     789                self.assertEqual(self.db.db, db.db)
     790                db = pg.DB(db=self.db.db)
     791                self.assertEqual(self.db.db, db.db)
     792                class DB2:
     793                        pass
     794                db2 = DB2()
     795                db2._cnx = self.db.db
     796                db = pg.DB(db2)
     797                self.assertEqual(self.db.db, db.db)
     798
    778799
    779800class TestDBClass(unittest.TestCase):
Note: See TracChangeset for help on using the changeset viewer.