psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-10-30 23:37:29 +03:00
Code Issues Packages Projects Releases Wiki Activity

Remove workarounds for namedtuple on Python <= 2.5

Browse Source 
namedtuple is available on all Python versions supported by psycopg2. It
was first introduced in Python 2.6. Can remove all workarounds and
special documentation.
This commit is contained in:
Jon Dufresne 2017-11-26 13:41:22 -08:00
parent 858bc3d42a
commit 05c28cce78
6 changed files with 11 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
doc/src/extras.rst
View File

@ -99,20 +99,6 @@ Real dictionary cursor
.. versionadded:: 2.3 .. versionadded:: 2.3
These objects require :py:func:`collections.namedtuple` to be found, so it is
available out-of-the-box only from Python 2.6. Anyway, the namedtuple
implementation is compatible with previous Python versions, so all you
have to do is to `download it`__ and make it available where we
expect it to be... ::
from somewhere import namedtuple
import collections
collections.namedtuple = namedtuple
from psycopg.extras import NamedTupleConnection
# ...
.. __: http://code.activestate.com/recipes/500261-named-tuples/
.. autoclass:: NamedTupleCursor .. autoclass:: NamedTupleCursor
.. autoclass:: NamedTupleConnection .. autoclass:: NamedTupleConnection

21
lib/extras.py
View File

@ -29,6 +29,7 @@ import os as _os
import sys as _sys import sys as _sys
import time as _time import time as _time
import re as _re import re as _re
from collections import namedtuple
try: try:
import logging as _logging import logging as _logging
@ -361,14 +362,8 @@ class NamedTupleCursor(_cursor):
except StopIteration: except StopIteration:
return return
try: def _make_nt(self):
from collections import namedtuple return namedtuple("Record", [d[0] for d in self.description or ()])
except ImportError as _exc:
def _make_nt(self):
raise self._exc
else:
def _make_nt(self, namedtuple=namedtuple):
return namedtuple("Record", [d[0] for d in self.description or ()])
class LoggingConnection(_connection): class LoggingConnection(_connection):
@ -1055,14 +1050,8 @@ class CompositeCaster(object):
return rv return rv
def _create_type(self, name, attnames): def _create_type(self, name, attnames):
try: self.type = namedtuple(name, attnames)
from collections import namedtuple self._ctor = self.type._make
except ImportError:
self.type = tuple
self._ctor = self.type
else:
self.type = namedtuple(name, attnames)
self._ctor = self.type._make
@classmethod @classmethod
def _from_db(self, name, conn_or_curs): def _from_db(self, name, conn_or_curs):

3
tests/test_cursor.py
View File

@ -27,7 +27,7 @@ import pickle
import psycopg2 import psycopg2
import psycopg2.extensions import psycopg2.extensions
from testutils import (unittest, ConnectingTestCase, skip_before_postgres, from testutils import (unittest, ConnectingTestCase, skip_before_postgres,
skip_if_no_namedtuple, skip_if_no_getrefcount, slow, skip_if_no_superuser, skip_if_no_getrefcount, slow, skip_if_no_superuser,
skip_if_windows) skip_if_windows)
import psycopg2.extras import psycopg2.extras
@ -377,7 +377,6 @@ class CursorTests(ConnectingTestCase):
for i, rec in enumerate(curs): for i, rec in enumerate(curs):
self.assertEqual(i + 1, curs.rownumber) self.assertEqual(i + 1, curs.rownumber)
@skip_if_no_namedtuple
def test_namedtuple_description(self): def test_namedtuple_description(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("""select curs.execute("""select

41
tests/test_extras_dictcursor.py
View File

@ -19,7 +19,6 @@ from datetime import timedelta
import psycopg2 import psycopg2
import psycopg2.extras import psycopg2.extras
from testutils import unittest, ConnectingTestCase, skip_before_postgres from testutils import unittest, ConnectingTestCase, skip_before_postgres
from testutils import skip_if_no_namedtuple
class ExtrasDictCursorTests(ConnectingTestCase): class ExtrasDictCursorTests(ConnectingTestCase):
@ -232,11 +231,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
ConnectingTestCase.setUp(self) ConnectingTestCase.setUp(self)
from psycopg2.extras import NamedTupleConnection from psycopg2.extras import NamedTupleConnection
try:
from collections import namedtuple # noqa
except ImportError:
return
self.conn = self.connect(connection_factory=NamedTupleConnection) self.conn = self.connect(connection_factory=NamedTupleConnection)
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("CREATE TEMPORARY TABLE nttest (i int, s text)") curs.execute("CREATE TEMPORARY TABLE nttest (i int, s text)")
@ -245,13 +239,11 @@ class NamedTupleCursorTest(ConnectingTestCase):
curs.execute("INSERT INTO nttest VALUES (3, 'baz')") curs.execute("INSERT INTO nttest VALUES (3, 'baz')")
self.conn.commit() self.conn.commit()
@skip_if_no_namedtuple
def test_cursor_args(self): def test_cursor_args(self):
cur = self.conn.cursor('foo', cursor_factory=psycopg2.extras.DictCursor) cur = self.conn.cursor('foo', cursor_factory=psycopg2.extras.DictCursor)
self.assertEqual(cur.name, 'foo') self.assertEqual(cur.name, 'foo')
self.assert_(isinstance(cur, psycopg2.extras.DictCursor)) self.assert_(isinstance(cur, psycopg2.extras.DictCursor))
@skip_if_no_namedtuple
def test_fetchone(self): def test_fetchone(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("select * from nttest order by 1") curs.execute("select * from nttest order by 1")
@ -263,7 +255,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 1) self.assertEqual(curs.rownumber, 1)
self.assertEqual(curs.rowcount, 3) self.assertEqual(curs.rowcount, 3)
@skip_if_no_namedtuple
def test_fetchmany_noarg(self): def test_fetchmany_noarg(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.arraysize = 2 curs.arraysize = 2
@ -277,7 +268,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 2) self.assertEqual(curs.rownumber, 2)
self.assertEqual(curs.rowcount, 3) self.assertEqual(curs.rowcount, 3)
@skip_if_no_namedtuple
def test_fetchmany(self): def test_fetchmany(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("select * from nttest order by 1") curs.execute("select * from nttest order by 1")
@ -290,7 +280,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 2) self.assertEqual(curs.rownumber, 2)
self.assertEqual(curs.rowcount, 3) self.assertEqual(curs.rowcount, 3)
@skip_if_no_namedtuple
def test_fetchall(self): def test_fetchall(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("select * from nttest order by 1") curs.execute("select * from nttest order by 1")
@ -305,7 +294,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 3) self.assertEqual(curs.rownumber, 3)
self.assertEqual(curs.rowcount, 3) self.assertEqual(curs.rowcount, 3)
@skip_if_no_namedtuple
def test_executemany(self): def test_executemany(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.executemany("delete from nttest where i = %s", curs.executemany("delete from nttest where i = %s",
@ -316,7 +304,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(res[0].i, 3) self.assertEqual(res[0].i, 3)
self.assertEqual(res[0].s, 'baz') self.assertEqual(res[0].s, 'baz')
@skip_if_no_namedtuple
def test_iter(self): def test_iter(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("select * from nttest order by 1") curs.execute("select * from nttest order by 1")
@ -342,26 +329,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 3) self.assertEqual(curs.rownumber, 3)
self.assertEqual(curs.rowcount, 3) self.assertEqual(curs.rowcount, 3)
def test_error_message(self):
try:
from collections import namedtuple # noqa
except ImportError:
# an import error somewhere
from psycopg2.extras import NamedTupleConnection
try:
self.conn = self.connect(
connection_factory=NamedTupleConnection)
curs = self.conn.cursor()
curs.execute("select 1")
curs.fetchone()
except ImportError:
pass
else:
self.fail("expecting ImportError")
else:
return self.skipTest("namedtuple available")
@skip_if_no_namedtuple
def test_record_updated(self): def test_record_updated(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("select 1 as foo;") curs.execute("select 1 as foo;")
@ -373,7 +340,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(r.bar, 2) self.assertEqual(r.bar, 2)
self.assertRaises(AttributeError, getattr, r, 'foo') self.assertRaises(AttributeError, getattr, r, 'foo')
@skip_if_no_namedtuple
def test_no_result_no_surprise(self): def test_no_result_no_surprise(self):
curs = self.conn.cursor() curs = self.conn.cursor()
curs.execute("update nttest set s = s") curs.execute("update nttest set s = s")
@ -382,7 +348,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
curs.execute("update nttest set s = s") curs.execute("update nttest set s = s")
self.assertRaises(psycopg2.ProgrammingError, curs.fetchall) self.assertRaises(psycopg2.ProgrammingError, curs.fetchall)
@skip_if_no_namedtuple
def test_minimal_generation(self): def test_minimal_generation(self):
# Instrument the class to verify it gets called the minimum number of times. # Instrument the class to verify it gets called the minimum number of times.
from psycopg2.extras import NamedTupleCursor from psycopg2.extras import NamedTupleCursor
@ -416,7 +381,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
finally: finally:
NamedTupleCursor._make_nt = f_orig NamedTupleCursor._make_nt = f_orig
@skip_if_no_namedtuple
@skip_before_postgres(8, 0) @skip_before_postgres(8, 0)
def test_named(self): def test_named(self):
curs = self.conn.cursor('tmp') curs = self.conn.cursor('tmp')
@ -427,28 +391,24 @@ class NamedTupleCursorTest(ConnectingTestCase):
recs.extend(curs.fetchall()) recs.extend(curs.fetchall())
self.assertEqual(range(10), [t.i for t in recs]) self.assertEqual(range(10), [t.i for t in recs])
@skip_if_no_namedtuple
def test_named_fetchone(self): def test_named_fetchone(self):
curs = self.conn.cursor('tmp') curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""") curs.execute("""select 42 as i""")
t = curs.fetchone() t = curs.fetchone()
self.assertEqual(t.i, 42) self.assertEqual(t.i, 42)
@skip_if_no_namedtuple
def test_named_fetchmany(self): def test_named_fetchmany(self):
curs = self.conn.cursor('tmp') curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""") curs.execute("""select 42 as i""")
recs = curs.fetchmany(10) recs = curs.fetchmany(10)
self.assertEqual(recs[0].i, 42) self.assertEqual(recs[0].i, 42)
@skip_if_no_namedtuple
def test_named_fetchall(self): def test_named_fetchall(self):
curs = self.conn.cursor('tmp') curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""") curs.execute("""select 42 as i""")
recs = curs.fetchall() recs = curs.fetchall()
self.assertEqual(recs[0].i, 42) self.assertEqual(recs[0].i, 42)
@skip_if_no_namedtuple
@skip_before_postgres(8, 2) @skip_before_postgres(8, 2)
def test_not_greedy(self): def test_not_greedy(self):
curs = self.conn.cursor('tmp') curs = self.conn.cursor('tmp')
@ -463,7 +423,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assert_(recs[1].ts - recs[0].ts < timedelta(seconds=0.005)) self.assert_(recs[1].ts - recs[0].ts < timedelta(seconds=0.005))
self.assert_(recs[2].ts - recs[1].ts > timedelta(seconds=0.0099)) self.assert_(recs[2].ts - recs[1].ts > timedelta(seconds=0.0099))
@skip_if_no_namedtuple
@skip_before_postgres(8, 0) @skip_before_postgres(8, 0)
def test_named_rownumber(self): def test_named_rownumber(self):
curs = self.conn.cursor('tmp') curs = self.conn.cursor('tmp')

22
tests/test_types_extras.py
View File

@ -541,16 +541,10 @@ class AdaptTypeTestCase(ConnectingTestCase):
self.assertEqual(v[0], 10) self.assertEqual(v[0], 10)
self.assertEqual(v[1], "hello") self.assertEqual(v[1], "hello")
self.assertEqual(v[2], date(2011, 1, 2)) self.assertEqual(v[2], date(2011, 1, 2))
self.assert_(t.type is not tuple)
try: self.assertEqual(v.anint, 10)
from collections import namedtuple # noqa self.assertEqual(v.astring, "hello")
except ImportError: self.assertEqual(v.adate, date(2011, 1, 2))
pass
else:
self.assert_(t.type is not tuple)
self.assertEqual(v.anint, 10)
self.assertEqual(v.astring, "hello")
self.assertEqual(v.adate, date(2011, 1, 2))
@skip_if_no_composite @skip_if_no_composite
def test_empty_string(self): def test_empty_string(self):
@ -591,13 +585,7 @@ class AdaptTypeTestCase(ConnectingTestCase):
v = curs.fetchone()[0] v = curs.fetchone()[0]
self.assertEqual(r, v) self.assertEqual(r, v)
self.assertEqual(v.anotherpair.apair.astring, "hello")
try:
from collections import namedtuple # noqa
except ImportError:
pass
else:
self.assertEqual(v.anotherpair.apair.astring, "hello")
@skip_if_no_composite @skip_if_no_composite
def test_register_on_cursor(self): def test_register_on_cursor(self):

13
tests/testutils.py
View File

@ -248,19 +248,6 @@ def skip_if_tpc_disabled(f):
return skip_if_tpc_disabled_ return skip_if_tpc_disabled_
def skip_if_no_namedtuple(f):
@wraps(f)
def skip_if_no_namedtuple_(self):
try:
from collections import namedtuple # noqa
except ImportError:
return self.skipTest("collections.namedtuple not available")
else:
return f(self)
return skip_if_no_namedtuple_
def skip_if_no_iobase(f): def skip_if_no_iobase(f):
"""Skip a test if io.TextIOBase is not available.""" """Skip a test if io.TextIOBase is not available."""
@wraps(f) @wraps(f)