我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用psycopg2.extras.register_composite()。
def test_empty_string(self): # issue #141 self._create_type("type_ss", [('s1', 'text'), ('s2', 'text')]) curs = self.conn.cursor() psycopg2.extras.register_composite("type_ss", curs) def ok(t): curs.execute("select %s::type_ss", (t,)) rv = curs.fetchone()[0] self.assertEqual(t, rv) ok(('a', 'b')) ok(('a', '')) ok(('', 'b')) ok(('a', None)) ok((None, 'b')) ok(('', '')) ok((None, None))
def test_cast_nested(self): self._create_type("type_is", [("anint", "integer"), ("astring", "text")]) self._create_type("type_r_dt", [("adate", "date"), ("apair", "type_is")]) self._create_type("type_r_ft", [("afloat", "float8"), ("anotherpair", "type_r_dt")]) psycopg2.extras.register_composite("type_is", self.conn) psycopg2.extras.register_composite("type_r_dt", self.conn) psycopg2.extras.register_composite("type_r_ft", self.conn) curs = self.conn.cursor() r = (0.25, (date(2011, 1, 2), (42, "hello"))) curs.execute("select %s::type_r_ft;", (r,)) 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")
def test_register_on_connection(self): self._create_type("type_ii", [("a", "integer"), ("b", "integer")]) conn1 = self.connect() conn2 = self.connect() try: psycopg2.extras.register_composite("type_ii", conn1) curs1 = conn1.cursor() curs2 = conn2.cursor() curs1.execute("select (1,2)::type_ii") self.assertEqual(curs1.fetchone()[0], (1, 2)) curs2.execute("select (1,2)::type_ii") self.assertEqual(curs2.fetchone()[0], "(1,2)") finally: conn1.close() conn2.close()
def test_composite_namespace(self): curs = self.conn.cursor() curs.execute(""" select nspname from pg_namespace where nspname = 'typens'; """) if not curs.fetchone(): curs.execute("create schema typens;") self.conn.commit() self._create_type("typens.typens_ii", [("a", "integer"), ("b", "integer")]) t = psycopg2.extras.register_composite( "typens.typens_ii", self.conn) self.assertEqual(t.schema, 'typens') curs.execute("select (4,8)::typens.typens_ii") self.assertEqual(curs.fetchone()[0], (4, 8))
def test_composite_array(self): self._create_type("type_isd", [('anint', 'integer'), ('astring', 'text'), ('adate', 'date')]) t = psycopg2.extras.register_composite("type_isd", self.conn) curs = self.conn.cursor() r1 = (10, 'hello', date(2011, 1, 2)) r2 = (20, 'world', date(2011, 1, 3)) curs.execute("select %s::type_isd[];", ([r1, r2],)) v = curs.fetchone()[0] self.assertEqual(len(v), 2) self.assertTrue(isinstance(v[0], t.type)) self.assertEqual(v[0][0], 10) self.assertEqual(v[0][1], "hello") self.assertEqual(v[0][2], date(2011, 1, 2)) self.assertTrue(isinstance(v[1], t.type)) self.assertEqual(v[1][0], 20) self.assertEqual(v[1][1], "world") self.assertEqual(v[1][2], date(2011, 1, 3))
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_subclass(self): oid = self._create_type("type_isd", [('anint', 'integer'), ('astring', 'text'), ('adate', 'date')]) from psycopg2.extras import register_composite, CompositeCaster class DictComposite(CompositeCaster): def make(self, values): return dict(list(zip(self.attnames, values))) t = register_composite('type_isd', self.conn, factory=DictComposite) self.assertEqual(t.name, 'type_isd') self.assertEqual(t.oid, oid) curs = self.conn.cursor() r = (10, 'hello', date(2011, 1, 2)) curs.execute("select %s::type_isd;", (r,)) v = curs.fetchone()[0] self.assertTrue(isinstance(v, dict)) self.assertEqual(v['anint'], 10) self.assertEqual(v['astring'], "hello") self.assertEqual(v['adate'], date(2011, 1, 2))
def test_register_globally(self): self._create_type("type_ii", [("a", "integer"), ("b", "integer")]) conn1 = self.connect() conn2 = self.connect() try: t = psycopg2.extras.register_composite("type_ii", conn1, globally=True) try: curs1 = conn1.cursor() curs2 = conn2.cursor() curs1.execute("select (1,2)::type_ii") self.assertEqual(curs1.fetchone()[0], (1, 2)) curs2.execute("select (1,2)::type_ii") self.assertEqual(curs2.fetchone()[0], (1, 2)) finally: # drop the registered typecasters to help the refcounting # script to return precise values. del psycopg2.extensions.string_types[t.typecaster.values[0]] if t.array_typecaster: del psycopg2.extensions.string_types[ t.array_typecaster.values[0]] finally: conn1.close() conn2.close()
def test_composite_array(self): self._create_type("type_isd", [('anint', 'integer'), ('astring', 'text'), ('adate', 'date')]) t = psycopg2.extras.register_composite("type_isd", self.conn) curs = self.conn.cursor() r1 = (10, 'hello', date(2011, 1, 2)) r2 = (20, 'world', date(2011, 1, 3)) curs.execute("select %s::type_isd[];", ([r1, r2],)) v = curs.fetchone()[0] self.assertEqual(len(v), 2) self.assert_(isinstance(v[0], t.type)) self.assertEqual(v[0][0], 10) self.assertEqual(v[0][1], "hello") self.assertEqual(v[0][2], date(2011, 1, 2)) self.assert_(isinstance(v[1], t.type)) self.assertEqual(v[1][0], 20) self.assertEqual(v[1][1], "world") self.assertEqual(v[1][2], date(2011, 1, 3))
def test_subclass(self): oid = self._create_type("type_isd", [('anint', 'integer'), ('astring', 'text'), ('adate', 'date')]) from psycopg2.extras import register_composite, CompositeCaster class DictComposite(CompositeCaster): def make(self, values): return dict(zip(self.attnames, values)) t = register_composite('type_isd', self.conn, factory=DictComposite) self.assertEqual(t.name, 'type_isd') self.assertEqual(t.oid, oid) curs = self.conn.cursor() r = (10, 'hello', date(2011, 1, 2)) curs.execute("select %s::type_isd;", (r,)) v = curs.fetchone()[0] self.assert_(isinstance(v, dict)) self.assertEqual(v['anint'], 10) self.assertEqual(v['astring'], "hello") self.assertEqual(v['adate'], date(2011, 1, 2))
def test_cast_nested(self): self._create_type("type_is", [("anint", "integer"), ("astring", "text")]) self._create_type("type_r_dt", [("adate", "date"), ("apair", "type_is")]) self._create_type("type_r_ft", [("afloat", "float8"), ("anotherpair", "type_r_dt")]) psycopg2.extras.register_composite("type_is", self.conn) psycopg2.extras.register_composite("type_r_dt", self.conn) psycopg2.extras.register_composite("type_r_ft", self.conn) curs = self.conn.cursor() r = (0.25, (date(2011,1,2), (42, "hello"))) curs.execute("select %s::type_r_ft;", (r,)) v = curs.fetchone()[0] self.assertEqual(r, v) try: from collections import namedtuple except ImportError: pass else: self.assertEqual(v.anotherpair.apair.astring, "hello")
def test_register_on_connection(self): self._create_type("type_ii", [("a", "integer"), ("b", "integer")]) conn1 = self.connect() conn2 = self.connect() try: psycopg2.extras.register_composite("type_ii", conn1) curs1 = conn1.cursor() curs2 = conn2.cursor() curs1.execute("select (1,2)::type_ii") self.assertEqual(curs1.fetchone()[0], (1,2)) curs2.execute("select (1,2)::type_ii") self.assertEqual(curs2.fetchone()[0], "(1,2)") finally: conn1.close() conn2.close()
def test_composite_namespace(self): curs = self.conn.cursor() curs.execute(""" select nspname from pg_namespace where nspname = 'typens'; """) if not curs.fetchone(): curs.execute("create schema typens;") self.conn.commit() self._create_type("typens.typens_ii", [("a", "integer"), ("b", "integer")]) t = psycopg2.extras.register_composite( "typens.typens_ii", self.conn) self.assertEqual(t.schema, 'typens') curs.execute("select (4,8)::typens.typens_ii") self.assertEqual(curs.fetchone()[0], (4,8))
def test_composite_array(self): oid = self._create_type("type_isd", [('anint', 'integer'), ('astring', 'text'), ('adate', 'date')]) t = psycopg2.extras.register_composite("type_isd", self.conn) curs = self.conn.cursor() r1 = (10, 'hello', date(2011,1,2)) r2 = (20, 'world', date(2011,1,3)) curs.execute("select %s::type_isd[];", ([r1, r2],)) v = curs.fetchone()[0] self.assertEqual(len(v), 2) self.assert_(isinstance(v[0], t.type)) self.assertEqual(v[0][0], 10) self.assertEqual(v[0][1], "hello") self.assertEqual(v[0][2], date(2011,1,2)) self.assert_(isinstance(v[1], t.type)) self.assertEqual(v[1][0], 20) self.assertEqual(v[1][1], "world") self.assertEqual(v[1][2], date(2011,1,3))
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_subclass(self): oid = self._create_type("type_isd", [('anint', 'integer'), ('astring', 'text'), ('adate', 'date')]) from psycopg2.extras import register_composite, CompositeCaster class DictComposite(CompositeCaster): def make(self, values): return dict(zip(self.attnames, values)) t = register_composite('type_isd', self.conn, factory=DictComposite) self.assertEqual(t.name, 'type_isd') self.assertEqual(t.oid, oid) curs = self.conn.cursor() r = (10, 'hello', date(2011,1,2)) curs.execute("select %s::type_isd;", (r,)) v = curs.fetchone()[0] self.assert_(isinstance(v, dict)) self.assertEqual(v['anint'], 10) self.assertEqual(v['astring'], "hello") self.assertEqual(v['adate'], date(2011,1,2))
def test_register_globally(self): self._create_type("type_ii", [("a", "integer"), ("b", "integer")]) conn1 = self.connect() conn2 = self.connect() try: t = psycopg2.extras.register_composite("type_ii", conn1, globally=True) try: curs1 = conn1.cursor() curs2 = conn2.cursor() curs1.execute("select (1,2)::type_ii") self.assertEqual(curs1.fetchone()[0], (1,2)) curs2.execute("select (1,2)::type_ii") self.assertEqual(curs2.fetchone()[0], (1,2)) finally: # drop the registered typecasters to help the refcounting # script to return precise values. del psycopg2.extensions.string_types[t.typecaster.values[0]] if t.array_typecaster: del psycopg2.extensions.string_types[ t.array_typecaster.values[0]] finally: conn1.close() conn2.close()
