mirror of
https://github.com/psycopg/psycopg2.git
synced 2025-01-31 17:34:08 +03:00
The hstore typecast can be registered globally.
This commit is contained in:
parent
bb41acd1da
commit
d5bf400cb4
|
@ -581,17 +581,10 @@ class HstoreAdapter(object):
|
||||||
|
|
||||||
parse_unicode = classmethod(parse_unicode)
|
parse_unicode = classmethod(parse_unicode)
|
||||||
|
|
||||||
def register_hstore(conn_or_curs, unicode=False):
|
def get_oids(self, conn_or_curs):
|
||||||
"""Register adapter/typecaster for dict/hstore reading/writing.
|
"""Return the oid of the hstore and hstore[] types.
|
||||||
|
|
||||||
The adapter must be registered on a connection or cursor as the hstore
|
Return None if hstore is not available.
|
||||||
oid is different in every database.
|
|
||||||
|
|
||||||
Raise `~psycopg2.ProgrammingError` if hstore is not installed in the
|
|
||||||
target database.
|
|
||||||
|
|
||||||
By default the returned dicts have string keys and values: use
|
|
||||||
*unicode*=True to return `unicode` objects instead.
|
|
||||||
"""
|
"""
|
||||||
if hasattr(conn_or_curs, 'execute'):
|
if hasattr(conn_or_curs, 'execute'):
|
||||||
conn = conn_or_curs.connection
|
conn = conn_or_curs.connection
|
||||||
|
@ -617,6 +610,26 @@ WHERE typname = 'hstore' and nspname = 'public';
|
||||||
and conn.isolation_level != _ext.ISOLATION_LEVEL_AUTOCOMMIT):
|
and conn.isolation_level != _ext.ISOLATION_LEVEL_AUTOCOMMIT):
|
||||||
conn.rollback()
|
conn.rollback()
|
||||||
|
|
||||||
|
return oids
|
||||||
|
|
||||||
|
get_oids = classmethod(get_oids)
|
||||||
|
|
||||||
|
def register_hstore(conn_or_curs, globally=False, unicode=False):
|
||||||
|
"""Register adapter/typecaster for dict/hstore reading/writing.
|
||||||
|
|
||||||
|
The function must receive a connection or cursor as the :sql:`hstore` oid
|
||||||
|
is different in every database. The typecaster will be registered only on
|
||||||
|
the connection or cursor passed as argument. If your application uses a
|
||||||
|
single database you can pass *globally*=True to have hstore registered on
|
||||||
|
all the connections.
|
||||||
|
|
||||||
|
Raise `~psycopg2.ProgrammingError` if hstore is not installed in the
|
||||||
|
target database.
|
||||||
|
|
||||||
|
By default the returned dicts have string keys and values: use
|
||||||
|
*unicode*=True to return `unicode` objects instead.
|
||||||
|
"""
|
||||||
|
oids = HstoreAdapter.get_oids(conn_or_curs)
|
||||||
if oids is None:
|
if oids is None:
|
||||||
raise psycopg2.ProgrammingError(
|
raise psycopg2.ProgrammingError(
|
||||||
"hstore type not found in the database. "
|
"hstore type not found in the database. "
|
||||||
|
@ -629,7 +642,7 @@ WHERE typname = 'hstore' and nspname = 'public';
|
||||||
cast = HstoreAdapter.parse
|
cast = HstoreAdapter.parse
|
||||||
|
|
||||||
HSTORE = _ext.new_type((oids[0],), "HSTORE", cast)
|
HSTORE = _ext.new_type((oids[0],), "HSTORE", cast)
|
||||||
_ext.register_type(HSTORE, conn_or_curs)
|
_ext.register_type(HSTORE, not globally and conn_or_curs or None)
|
||||||
_ext.register_adapter(dict, HstoreAdapter)
|
_ext.register_adapter(dict, HstoreAdapter)
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -222,6 +222,27 @@ class HstoreTestCase(unittest.TestCase):
|
||||||
self.assert_(isinstance(t[2].keys()[0], unicode))
|
self.assert_(isinstance(t[2].keys()[0], unicode))
|
||||||
self.assert_(isinstance(t[2].values()[0], unicode))
|
self.assert_(isinstance(t[2].values()[0], unicode))
|
||||||
|
|
||||||
|
def test_register_globally(self):
|
||||||
|
from psycopg2.extras import register_hstore, HstoreAdapter
|
||||||
|
|
||||||
|
oids = HstoreAdapter.get_oids(self.conn)
|
||||||
|
try:
|
||||||
|
register_hstore(self.conn, globally=True)
|
||||||
|
conn2 = psycopg2.connect(self.conn.dsn)
|
||||||
|
cur2 = self.conn.cursor()
|
||||||
|
cur2.execute("select 'a => b'::hstore")
|
||||||
|
r = cur2.fetchone()
|
||||||
|
self.assert_(isinstance(r[0], dict))
|
||||||
|
conn2.close()
|
||||||
|
finally:
|
||||||
|
psycopg2.extensions.string_types.pop(oids[0])
|
||||||
|
|
||||||
|
# verify the caster is not around anymore
|
||||||
|
cur = self.conn.cursor()
|
||||||
|
cur.execute("select 'a => b'::hstore")
|
||||||
|
r = cur.fetchone()
|
||||||
|
self.assert_(isinstance(r[0], str))
|
||||||
|
|
||||||
def test_roundtrip(self):
|
def test_roundtrip(self):
|
||||||
from psycopg2.extras import register_hstore
|
from psycopg2.extras import register_hstore
|
||||||
register_hstore(self.conn)
|
register_hstore(self.conn)
|
||||||
|
|
Loading…
Reference in New Issue
Block a user