def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
python类RealDictConnection()的实例源码
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
test_types_extras.py 文件源码
项目：aws-lambda-redshift-copy
作者: christianhxc
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
test_types_extras.py 文件源码
项目：aws-lambda-redshift-copy
作者: christianhxc
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1,2))
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_hstore
conn = self.connect(connection_factory=RealDictConnection)
try:
register_hstore(conn)
curs = conn.cursor()
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_hstore(curs)
curs.execute("select ''::hstore as x")
self.assertEqual(curs.fetchone()['x'], {})
finally:
conn.close()
def test_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
from psycopg2.extras import register_composite
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
conn = self.connect(connection_factory=RealDictConnection)
try:
register_composite('type_ii', conn)
curs = conn.cursor()
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
conn = self.connect(connection_factory=RealDictConnection)
try:
curs = conn.cursor()
register_composite('type_ii', conn)
curs.execute("select '(1,2)'::type_ii as x")
self.assertEqual(curs.fetchone()['x'], (1, 2))
finally:
conn.close()
def test_tpc_recover_non_dbapi_connection(self):
from psycopg2.extras import RealDictConnection
cnn = self.connect(connection_factory=RealDictConnection)
cnn.tpc_begin('dict-connection')
cnn.tpc_prepare()
cnn.reset()
xids = cnn.tpc_recover()
xid = [x for x in xids if x.database == dbname][0]
self.assertEqual(None, xid.format_id)
self.assertEqual('dict-connection', xid.gtrid)
self.assertEqual(None, xid.bqual)
