Added support for inet array

This commit is contained in:
Daniele Varrazzo 2012-02-23 23:56:55 +00:00
parent 36b6c80ed1
commit 0c337a2029
4 changed files with 37 additions and 7 deletions

1
NEWS
View File

@ -11,6 +11,7 @@ What's new in psycopg 2.4.5
to Menno Smits (tickets #94, #95).
- Fixed 'rownumber' during iteration on cursor subclasses.
Regression introduced in 2.4.4 (ticket #100).
- Added support for 'inet' arrays.
What's new in psycopg 2.4.4

View File

@ -250,6 +250,7 @@ UUID data type
^^^^^^^^^^^^^^^^^^^^^^
.. versionadded:: 2.0.9
.. versionchanged:: 2.4.5 added inet array support.
.. doctest::
@ -264,7 +265,7 @@ UUID data type
'192.168.0.1/24'
.. autofunction:: register_inet()
.. autofunction:: register_inet
.. autoclass:: Inet

View File

@ -498,13 +498,13 @@ class Inet(object):
"""
def __init__(self, addr):
self.addr = addr
def __repr__(self):
return "%s(%r)" % (self.__class__.__name__, self.addr)
def prepare(self, conn):
self._conn = conn
def getquoted(self):
obj = _A(self.addr)
if hasattr(obj, 'prepare'):
@ -517,13 +517,32 @@ class Inet(object):
def __str__(self):
return str(self.addr)
def register_inet(oid=None, conn_or_curs=None):
"""Create the INET type and an Inet adapter."""
if not oid: oid = 869
_ext.INET = _ext.new_type((oid, ), "INET",
"""Create the INET type and an Inet adapter.
:param oid: oid for the PostgreSQL :sql:`inet` type, or 2-items sequence
with oids of the type and the array. If not specified, use PostgreSQL
standard oids.
:param conn_or_curs: where to register the typecaster. If not specified,
register it globally.
"""
if not oid:
oid1 = 869
oid2 = 1041
elif isinstance(oid, (list, tuple)):
oid1, oid2 = oid
else:
oid1 = oid
oid2 = 1041
_ext.INET = _ext.new_type((oid1, ), "INET",
lambda data, cursor: data and Inet(data) or None)
_ext.INETARRAY = _ext.new_array_type((oid2, ), "INETARRAY", _ext.INET)
_ext.register_type(_ext.INET, conn_or_curs)
_ext.register_type(_ext.INETARRAY, conn_or_curs)
return _ext.INET

View File

@ -89,6 +89,15 @@ class TypesExtrasTests(unittest.TestCase):
s = self.execute("SELECT NULL::inet AS foo")
self.failUnless(s is None)
def testINETARRAY(self):
psycopg2.extras.register_inet()
i = psycopg2.extras.Inet("192.168.1.0/24")
s = self.execute("SELECT %s AS foo", ([i],))
self.failUnless(i.addr == s[0].addr)
# must survive NULL cast to inet
s = self.execute("SELECT NULL::inet[] AS foo")
self.failUnless(s is None)
def test_inet_conform(self):
from psycopg2.extras import Inet
i = Inet("192.168.1.0/24")