psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-22 17:06:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Some light cleanup for Py3 conversion.

Browse Source 
Either flagged as warning by python2.6 -3 or converted by 2to3.
This commit is contained in:
Daniele Varrazzo 2010-12-12 16:45:21 +00:00
parent a30e461038
commit 31093a7a58
5 changed files with 33 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
lib/extras.py
View File

@ -26,6 +26,7 @@ and classes untill a better place in the distribution is found.
# License for more details. # License for more details.
import os import os
import sys
import time import time
import codecs import codecs
import warnings import warnings
@ -47,7 +48,7 @@ class DictCursorBase(_cursor):
"""Base class for all dict-like cursors.""" """Base class for all dict-like cursors."""
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
if kwargs.has_key('row_factory'): if 'row_factory' in kwargs:
row_factory = kwargs['row_factory'] row_factory = kwargs['row_factory']
del kwargs['row_factory'] del kwargs['row_factory']
else: else:
@ -140,20 +141,17 @@ class DictRow(list):
self[:] = [None] * len(cursor.description) self[:] = [None] * len(cursor.description)
def __getitem__(self, x): def __getitem__(self, x):
if type(x) != int: if not isinstance(x, int):
x = self._index[x] x = self._index[x]
return list.__getitem__(self, x) return list.__getitem__(self, x)
def __setitem__(self, x, v): def __setitem__(self, x, v):
if type(x) != int: if not isinstance(x, int):
x = self._index[x] x = self._index[x]
list.__setitem__(self, x, v) list.__setitem__(self, x, v)
def items(self): def items(self):
res = [] return list(self.iteritems())
for n, v in self._index.items():
res.append((n, list.__getitem__(self, v)))
return res
def keys(self): def keys(self):
return self._index.keys() return self._index.keys()
@ -162,7 +160,7 @@ class DictRow(list):
return tuple(self[:]) return tuple(self[:])
def has_key(self, x): def has_key(self, x):
return self._index.has_key(x) return x in self._index
def get(self, x, default=None): def get(self, x, default=None):
try: try:
@ -171,7 +169,7 @@ class DictRow(list):
return default return default
def iteritems(self): def iteritems(self):
for n, v in self._index.items(): for n, v in self._index.iteritems():
yield n, list.__getitem__(self, v) yield n, list.__getitem__(self, v)
def iterkeys(self): def iterkeys(self):
@ -181,10 +179,18 @@ class DictRow(list):
return list.__iter__(self) return list.__iter__(self)
def copy(self): def copy(self):
return dict(self.items()) return dict(self.iteritems())
def __contains__(self, x): def __contains__(self, x):
return self._index.__contains__(x) return x in self._index
# grop the crusty Py2 methods
if sys.version_info[0] > 2:
items = iteritems; del iteritems
keys = iterkeys; del iterkeys
values = itervalues; del itervalues
del has_key
class RealDictConnection(_connection): class RealDictConnection(_connection):
"""A connection that uses `RealDictCursor` automatically.""" """A connection that uses `RealDictCursor` automatically."""
@ -615,7 +621,10 @@ class HstoreAdapter(object):
""", regex.VERBOSE) """, regex.VERBOSE)
# backslash decoder # backslash decoder
_bsdec = codecs.getdecoder("string_escape") if sys.version_info[0] < 3:
_bsdec = codecs.getdecoder("string_escape")
else:
_bsdec = codecs.getdecoder("unicode_escape")
def parse(self, s, cur, _decoder=_bsdec): def parse(self, s, cur, _decoder=_bsdec):
"""Parse an hstore representation in a Python string. """Parse an hstore representation in a Python string.

2
lib/pool.py
View File

@ -100,7 +100,7 @@ class AbstractConnectionPool(object):
if self.closed: raise PoolError("connection pool is closed") if self.closed: raise PoolError("connection pool is closed")
if key is None: key = self._getkey() if key is None: key = self._getkey()
if self._used.has_key(key): if key in self._used:
return self._used[key] return self._used[key]
if self._pool: if self._pool:

2
tests/test_cursor.py
View File

@ -18,7 +18,7 @@ class CursorTests(unittest.TestCase):
cur = conn.cursor() cur = conn.cursor()
cur.execute("create temp table test_exc (data int);") cur.execute("create temp table test_exc (data int);")
def buggygen(): def buggygen():
yield 1/0 yield 1//0
self.assertRaises(ZeroDivisionError, self.assertRaises(ZeroDivisionError,
cur.executemany, "insert into test_exc values (%s)", buggygen()) cur.executemany, "insert into test_exc values (%s)", buggygen())
cur.close() cur.close()

18
tests/test_dates.py
View File

@ -243,18 +243,18 @@ class DatetimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
def test_type_roundtrip_date(self): def test_type_roundtrip_date(self):
from datetime import date from datetime import date
self._test_type_roundtrip(date(2010,05,03)) self._test_type_roundtrip(date(2010,5,3))
def test_type_roundtrip_datetime(self): def test_type_roundtrip_datetime(self):
from datetime import datetime from datetime import datetime
dt = self._test_type_roundtrip(datetime(2010,05,03,10,20,30)) dt = self._test_type_roundtrip(datetime(2010,5,3,10,20,30))
self.assertEqual(None, dt.tzinfo) self.assertEqual(None, dt.tzinfo)
def test_type_roundtrip_datetimetz(self): def test_type_roundtrip_datetimetz(self):
from datetime import datetime from datetime import datetime
import psycopg2.tz import psycopg2.tz
tz = psycopg2.tz.FixedOffsetTimezone(8*60) tz = psycopg2.tz.FixedOffsetTimezone(8*60)
dt1 = datetime(2010,05,03,10,20,30, tzinfo=tz) dt1 = datetime(2010,5,3,10,20,30, tzinfo=tz)
dt2 = self._test_type_roundtrip(dt1) dt2 = self._test_type_roundtrip(dt1)
self.assertNotEqual(None, dt2.tzinfo) self.assertNotEqual(None, dt2.tzinfo)
self.assertEqual(dt1, dt2) self.assertEqual(dt1, dt2)
@ -269,11 +269,11 @@ class DatetimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
def test_type_roundtrip_date_array(self): def test_type_roundtrip_date_array(self):
from datetime import date from datetime import date
self._test_type_roundtrip_array(date(2010,05,03)) self._test_type_roundtrip_array(date(2010,5,3))
def test_type_roundtrip_datetime_array(self): def test_type_roundtrip_datetime_array(self):
from datetime import datetime from datetime import datetime
self._test_type_roundtrip_array(datetime(2010,05,03,10,20,30)) self._test_type_roundtrip_array(datetime(2010,5,3,10,20,30))
def test_type_roundtrip_time_array(self): def test_type_roundtrip_time_array(self):
from datetime import time from datetime import time
@ -426,11 +426,11 @@ class mxDateTimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
def test_type_roundtrip_date(self): def test_type_roundtrip_date(self):
from mx.DateTime import Date from mx.DateTime import Date
self._test_type_roundtrip(Date(2010,05,03)) self._test_type_roundtrip(Date(2010,5,3))
def test_type_roundtrip_datetime(self): def test_type_roundtrip_datetime(self):
from mx.DateTime import DateTime from mx.DateTime import DateTime
self._test_type_roundtrip(DateTime(2010,05,03,10,20,30)) self._test_type_roundtrip(DateTime(2010,5,3,10,20,30))
def test_type_roundtrip_time(self): def test_type_roundtrip_time(self):
from mx.DateTime import Time from mx.DateTime import Time
@ -442,11 +442,11 @@ class mxDateTimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
def test_type_roundtrip_date_array(self): def test_type_roundtrip_date_array(self):
from mx.DateTime import Date from mx.DateTime import Date
self._test_type_roundtrip_array(Date(2010,05,03)) self._test_type_roundtrip_array(Date(2010,5,3))
def test_type_roundtrip_datetime_array(self): def test_type_roundtrip_datetime_array(self):
from mx.DateTime import DateTime from mx.DateTime import DateTime
self._test_type_roundtrip_array(DateTime(2010,05,03,10,20,30)) self._test_type_roundtrip_array(DateTime(2010,5,3,10,20,30))
def test_type_roundtrip_time_array(self): def test_type_roundtrip_time_array(self):
from mx.DateTime import Time from mx.DateTime import Time

2
tests/test_green.py
View File

@ -63,7 +63,7 @@ class GreenTests(unittest.TestCase):
curs.fetchone() curs.fetchone()
# now try to do something that will fail in the callback # now try to do something that will fail in the callback
psycopg2.extensions.set_wait_callback(lambda conn: 1/0) psycopg2.extensions.set_wait_callback(lambda conn: 1//0)
self.assertRaises(ZeroDivisionError, curs.execute, "select 2") self.assertRaises(ZeroDivisionError, curs.execute, "select 2")
# check that the connection is left in an usable state # check that the connection is left in an usable state