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

Merge remote-tracking branch 'jdufresne/namedtuple'

Browse Source
This commit is contained in:
Daniele Varrazzo 2017-11-28 03:02:56 +00:00
commit 7855f28785
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
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

21
lib/extras.py
View File

@ -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):

3
tests/test_cursor.py
View File

@ -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

41
tests/test_extras_dictcursor.py
View File

@ -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')

22
tests/test_types_extras.py
View File

@ -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):

13
tests/testutils.py
View File

@ -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)