Python psycopg2 模块,sql() 实例源码
我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用psycopg2.sql()。
def test_execute(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key,
foo text,bar text,"ba'z" text)
""")
cur.execute(
sql.sql("insert into {0} (id,{1}) values (%s,{2})").format(
sql.Identifier('test_compose'),
sql.sql(',').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),
(sql.Placeholder() * 3).join(',')),
(10, 'a', 'b', 'c'))
cur.execute("select * from test_compose")
self.assertEqual(cur.fetchall(), [(10, 'c')])
def test_executemany(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key,"ba'z" text)
""")
cur.executemany(
sql.sql("insert into {0} (id,
[(10, 'c'), (20, 'd', 'e', 'f')])
cur.execute("select * from test_compose")
self.assertEqual(cur.fetchall(), 'f')])
def test_execute(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'c')])
def test_executemany(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'f')])
def test_execute(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'c')])
def test_executemany(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'f')])
def test_copy(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key,"ba'z" text)
""")
s = StringIO("10\ta\tb\tc\n20\td\te\tf\n")
cur.copy_expert(
sql.sql("copy {t} (id,foo,bar,{f}) from stdin").format(
t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s)
s1 = StringIO()
cur.copy_expert(
sql.sql("copy (select {f} from {t} order by id) to stdout").format(
t=sql.Identifier("test_compose"), s1)
s1.seek(0)
self.assertEqual(s1.read(), 'c\nf\n')
def test_execute(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'c')])
def test_executemany(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'f')])
def test_execute(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'c')])
def test_executemany(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'f')])
def test_copy(self):
cur = self.conn.cursor()
cur.execute("""
create table test_compose (
id serial primary key, 'c\nf\n')
def test_pos(self):
s = sql.sql("select {} from {}").format(
sql.Identifier('field'), sql.Identifier('table'))
s1 = s.as_string(self.conn)
self.assertTrue(isinstance(s1, str))
self.assertEqual(s1, 'select "field" from "table"')
def test_pos_spec(self):
s = sql.sql("select {0} from {1}").format(
sql.Identifier('field'), 'select "field" from "table"')
s = sql.sql("select {1} from {0}").format(
sql.Identifier('table'), sql.Identifier('field'))
s1 = s.as_string(self.conn)
self.assertTrue(isinstance(s1, 'select "field" from "table"')
def test_dict(self):
s = sql.sql("select {f} from {t}").format(
f=sql.Identifier('field'), t=sql.Identifier('table'))
s1 = s.as_string(self.conn)
self.assertTrue(isinstance(s1, 'select "field" from "table"')
def test_string(self):
self.assertEqual(sql.Identifier('foo').string, 'foo')
def test_repr(self):
obj = sql.Identifier("fo'o")
self.assertEqual(repr(obj), 'Identifier("fo\'o")')
self.assertEqual(repr(obj), str(obj))
def test_join(self):
self.assertTrue(not hasattr(sql.Identifier('foo'), 'join'))
def test_init(self):
self.assertTrue(isinstance(sql.Literal('foo'), sql.Literal))
self.assertTrue(isinstance(sql.Literal('foo'), sql.Literal))
self.assertTrue(isinstance(sql.Literal(b'foo'), sql.Literal))
self.assertTrue(isinstance(sql.Literal(42), sql.Literal))
self.assertTrue(isinstance(
sql.Literal(dt.date(2016, 31)), sql.Literal))
def test_repr(self):
self.assertEqual(repr(sql.Literal("foo")), "Literal('foo')")
self.assertEqual(str(sql.Literal("foo")), "Literal('foo')")
self.assertQuotedEqual(sql.Literal("foo").as_string(self.conn), "'foo'")
self.assertEqual(sql.Literal(42).as_string(self.conn), "42")
self.assertEqual(
sql.Literal(dt.date(2017, 1, 1)).as_string(self.conn),
"'2017-01-01'::date")
def test_multiply(self):
obj = sql.sql("foo") * 3
self.assertTrue(isinstance(obj, "foofoofoo")
def test_join(self):
obj = sql.sql(",").join(
[sql.Identifier('foo'), sql.sql('bar'), sql.Literal(42)])
self.assertTrue(isinstance(obj, '"foo",42')
obj = sql.sql(",").join(
sql.Composed([sql.Identifier('foo'), sql.Literal(42)]))
self.assertTrue(isinstance(obj,").join([])
self.assertEqual(obj, sql.Composed([]))
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。