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

Added support for UUID arrays

Browse Source
This commit is contained in:
Federico Di Gregorio 2009-10-04 12:34:02 +02:00
parent 7022269b3d
commit 390a9c2451
3 changed files with 47 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ChangeLog
View File

@ -1,5 +1,7 @@
2009-10-04 Federico Di Gregorio <fog@initd.org>
* lib/extras.py: added support for UUID arrays.
* psycopg/connection_int.c: applied patch from Richard Davies to avoid
deadlocks with multiple threads using the same connection.

27
lib/extras.py
View File

@ -340,13 +340,34 @@ try:
__str__ = getquoted
def register_uuid(oid=None, conn_or_curs=None):
def register_uuid(oids=None, conn_or_curs=None):
"""Create the UUID type and an uuid.UUID adapter."""
if not oid: oid = 2950
_ext.UUID = _ext.new_type((oid, ), "UUID",
if not oids:
oid1 = 2950
oid2 = 2951
elif type(oids) == list:
oid1, oid2 = oids
else:
oid1 = oids
oid2 = 2951
def parseUUIDARRAY(data, cursor):
if data is None:
return None
elif data == '{}':
return []
else:
return [((len(x) > 0 and x != 'NULL') and uuid.UUID(x) or None)
for x in data[1:-1].split(',')]
_ext.UUID = _ext.new_type((oid1, ), "UUID",
lambda data, cursor: data and uuid.UUID(data) or None)
_ext.UUIDARRAY = _ext.new_type((oid2,), "UUID[]", parseUUIDARRAY)
_ext.register_type(_ext.UUID, conn_or_curs)
_ext.register_type(_ext.UUIDARRAY, conn_or_curs)
_ext.register_adapter(uuid.UUID, UUID_adapter)
return _ext.UUID
except ImportError, e:

22
tests/types_extras.py
View File

@ -42,13 +42,33 @@ class TypesExtrasTests(unittest.TestCase):
psycopg2.extras.register_uuid()
except:
return
u = uuid.UUID('9c6d5a77-7256-457e-9461-347b4358e350');
u = uuid.UUID('9c6d5a77-7256-457e-9461-347b4358e350')
s = self.execute("SELECT %s AS foo", (u,))
self.failUnless(u == s)
# must survive NULL cast to a uuid
s = self.execute("SELECT NULL::uuid AS foo")
self.failUnless(s is None)
def testUUIDARRAY(self):
try:
import uuid
psycopg2.extras.register_uuid()
except:
return
u = [uuid.UUID('9c6d5a77-7256-457e-9461-347b4358e350'), uuid.UUID('9c6d5a77-7256-457e-9461-347b4358e352')]
s = self.execute("SELECT %s AS foo", (u,))
self.failUnless(u == s)
# array with a NULL element
u = [uuid.UUID('9c6d5a77-7256-457e-9461-347b4358e350'), None]
s = self.execute("SELECT %s AS foo", (u,))
self.failUnless(u == s)
# must survive NULL cast to a uuid[]
s = self.execute("SELECT NULL::uuid[] AS foo")
self.failUnless(s is None)
# what about empty arrays?
s = self.execute("SELECT '{}'::uuid[] AS foo")
self.failUnless(type(s) == list and len(s) == 0)
def testINET(self):
psycopg2.extras.register_inet()
i = "192.168.1.0/24";