From 05c28cce78aa223e632635af813cbbaed4c82169 Mon Sep 17 00:00:00 2001 From: Jon Dufresne Date: Sun, 26 Nov 2017 13:41:22 -0800 Subject: [PATCH] Remove workarounds for namedtuple on Python <= 2.5 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. --- doc/src/extras.rst | 14 ----------- lib/extras.py | 21 ++++------------- tests/test_cursor.py | 3 +-- tests/test_extras_dictcursor.py | 41 --------------------------------- tests/test_types_extras.py | 22 ++++-------------- tests/testutils.py | 13 ----------- 6 files changed, 11 insertions(+), 103 deletions(-) diff --git a/doc/src/extras.rst b/doc/src/extras.rst index c844873a..cb8335be 100644 --- a/doc/src/extras.rst +++ b/doc/src/extras.rst @@ -99,20 +99,6 @@ Real dictionary cursor .. 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:: NamedTupleConnection diff --git a/lib/extras.py b/lib/extras.py index 8abb14fc..bdd61f7a 100644 --- a/lib/extras.py +++ b/lib/extras.py @@ -29,6 +29,7 @@ import os as _os import sys as _sys import time as _time import re as _re +from collections import namedtuple try: import logging as _logging @@ -361,14 +362,8 @@ class NamedTupleCursor(_cursor): except StopIteration: return - try: - from collections import namedtuple - 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 ()]) + def _make_nt(self): + return namedtuple("Record", [d[0] for d in self.description or ()]) class LoggingConnection(_connection): @@ -1055,14 +1050,8 @@ class CompositeCaster(object): return rv def _create_type(self, name, attnames): - try: - from collections import namedtuple - except ImportError: - self.type = tuple - self._ctor = self.type - else: - self.type = namedtuple(name, attnames) - self._ctor = self.type._make + self.type = namedtuple(name, attnames) + self._ctor = self.type._make @classmethod def _from_db(self, name, conn_or_curs): diff --git a/tests/test_cursor.py b/tests/test_cursor.py index 10b8d71e..dbc72e19 100755 --- a/tests/test_cursor.py +++ b/tests/test_cursor.py @@ -27,7 +27,7 @@ import pickle import psycopg2 import psycopg2.extensions 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) import psycopg2.extras @@ -377,7 +377,6 @@ class CursorTests(ConnectingTestCase): for i, rec in enumerate(curs): self.assertEqual(i + 1, curs.rownumber) - @skip_if_no_namedtuple def test_namedtuple_description(self): curs = self.conn.cursor() curs.execute("""select diff --git a/tests/test_extras_dictcursor.py b/tests/test_extras_dictcursor.py index 20393c66..2c867d38 100755 --- a/tests/test_extras_dictcursor.py +++ b/tests/test_extras_dictcursor.py @@ -19,7 +19,6 @@ from datetime import timedelta import psycopg2 import psycopg2.extras from testutils import unittest, ConnectingTestCase, skip_before_postgres -from testutils import skip_if_no_namedtuple class ExtrasDictCursorTests(ConnectingTestCase): @@ -232,11 +231,6 @@ class NamedTupleCursorTest(ConnectingTestCase): ConnectingTestCase.setUp(self) from psycopg2.extras import NamedTupleConnection - try: - from collections import namedtuple # noqa - except ImportError: - return - self.conn = self.connect(connection_factory=NamedTupleConnection) curs = self.conn.cursor() 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')") self.conn.commit() - @skip_if_no_namedtuple def test_cursor_args(self): cur = self.conn.cursor('foo', cursor_factory=psycopg2.extras.DictCursor) self.assertEqual(cur.name, 'foo') self.assert_(isinstance(cur, psycopg2.extras.DictCursor)) - @skip_if_no_namedtuple def test_fetchone(self): curs = self.conn.cursor() curs.execute("select * from nttest order by 1") @@ -263,7 +255,6 @@ class NamedTupleCursorTest(ConnectingTestCase): self.assertEqual(curs.rownumber, 1) self.assertEqual(curs.rowcount, 3) - @skip_if_no_namedtuple def test_fetchmany_noarg(self): curs = self.conn.cursor() curs.arraysize = 2 @@ -277,7 +268,6 @@ class NamedTupleCursorTest(ConnectingTestCase): self.assertEqual(curs.rownumber, 2) self.assertEqual(curs.rowcount, 3) - @skip_if_no_namedtuple def test_fetchmany(self): curs = self.conn.cursor() curs.execute("select * from nttest order by 1") @@ -290,7 +280,6 @@ class NamedTupleCursorTest(ConnectingTestCase): self.assertEqual(curs.rownumber, 2) self.assertEqual(curs.rowcount, 3) - @skip_if_no_namedtuple def test_fetchall(self): curs = self.conn.cursor() curs.execute("select * from nttest order by 1") @@ -305,7 +294,6 @@ class NamedTupleCursorTest(ConnectingTestCase): self.assertEqual(curs.rownumber, 3) self.assertEqual(curs.rowcount, 3) - @skip_if_no_namedtuple def test_executemany(self): curs = self.conn.cursor() 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].s, 'baz') - @skip_if_no_namedtuple def test_iter(self): curs = self.conn.cursor() curs.execute("select * from nttest order by 1") @@ -342,26 +329,6 @@ class NamedTupleCursorTest(ConnectingTestCase): self.assertEqual(curs.rownumber, 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): curs = self.conn.cursor() curs.execute("select 1 as foo;") @@ -373,7 +340,6 @@ class NamedTupleCursorTest(ConnectingTestCase): self.assertEqual(r.bar, 2) self.assertRaises(AttributeError, getattr, r, 'foo') - @skip_if_no_namedtuple def test_no_result_no_surprise(self): curs = self.conn.cursor() curs.execute("update nttest set s = s") @@ -382,7 +348,6 @@ class NamedTupleCursorTest(ConnectingTestCase): curs.execute("update nttest set s = s") self.assertRaises(psycopg2.ProgrammingError, curs.fetchall) - @skip_if_no_namedtuple def test_minimal_generation(self): # Instrument the class to verify it gets called the minimum number of times. from psycopg2.extras import NamedTupleCursor @@ -416,7 +381,6 @@ class NamedTupleCursorTest(ConnectingTestCase): finally: NamedTupleCursor._make_nt = f_orig - @skip_if_no_namedtuple @skip_before_postgres(8, 0) def test_named(self): curs = self.conn.cursor('tmp') @@ -427,28 +391,24 @@ class NamedTupleCursorTest(ConnectingTestCase): recs.extend(curs.fetchall()) self.assertEqual(range(10), [t.i for t in recs]) - @skip_if_no_namedtuple def test_named_fetchone(self): curs = self.conn.cursor('tmp') curs.execute("""select 42 as i""") t = curs.fetchone() self.assertEqual(t.i, 42) - @skip_if_no_namedtuple def test_named_fetchmany(self): curs = self.conn.cursor('tmp') curs.execute("""select 42 as i""") recs = curs.fetchmany(10) self.assertEqual(recs[0].i, 42) - @skip_if_no_namedtuple def test_named_fetchall(self): curs = self.conn.cursor('tmp') curs.execute("""select 42 as i""") recs = curs.fetchall() self.assertEqual(recs[0].i, 42) - @skip_if_no_namedtuple @skip_before_postgres(8, 2) def test_not_greedy(self): 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[2].ts - recs[1].ts > timedelta(seconds=0.0099)) - @skip_if_no_namedtuple @skip_before_postgres(8, 0) def test_named_rownumber(self): curs = self.conn.cursor('tmp') diff --git a/tests/test_types_extras.py b/tests/test_types_extras.py index e1db2c2c..a5aa7d2b 100755 --- a/tests/test_types_extras.py +++ b/tests/test_types_extras.py @@ -541,16 +541,10 @@ class AdaptTypeTestCase(ConnectingTestCase): self.assertEqual(v[0], 10) self.assertEqual(v[1], "hello") self.assertEqual(v[2], date(2011, 1, 2)) - - try: - from collections import namedtuple # noqa - except ImportError: - 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)) + 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 def test_empty_string(self): @@ -591,13 +585,7 @@ class AdaptTypeTestCase(ConnectingTestCase): v = curs.fetchone()[0] self.assertEqual(r, v) - - try: - from collections import namedtuple # noqa - except ImportError: - pass - else: - self.assertEqual(v.anotherpair.apair.astring, "hello") + self.assertEqual(v.anotherpair.apair.astring, "hello") @skip_if_no_composite def test_register_on_cursor(self): diff --git a/tests/testutils.py b/tests/testutils.py index 5c192e30..4e1ee37f 100644 --- a/tests/testutils.py +++ b/tests/testutils.py @@ -248,19 +248,6 @@ def skip_if_tpc_disabled(f): 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): """Skip a test if io.TextIOBase is not available.""" @wraps(f)