本文整理汇总了Python中sqlalchemy.dialects.postgresql.base.dialect函数的典型用法代码示例。如果您正苦于以下问题:Python dialect函数的具体用法?Python dialect怎么用?Python dialect使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了dialect函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_load_dialect_impl_sqlite
def test_load_dialect_impl_sqlite(self):
from sqlalchemy.dialects.sqlite.base import dialect
from sqlalchemy.types import CHAR
inet = self.INET()
impl = inet.load_dialect_impl(dialect())
self.assertTrue(isinstance(impl, CHAR))
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:7,代码来源:test_inet.py
示例2: test_serial_integer
def test_serial_integer(self):
class BITD(TypeDecorator):
impl = Integer
def load_dialect_impl(self, dialect):
if dialect.name == "postgresql":
return BigInteger()
else:
return Integer()
for version, type_, expected in [
(None, Integer, "SERIAL"),
(None, BigInteger, "BIGSERIAL"),
((9, 1), SmallInteger, "SMALLINT"),
((9, 2), SmallInteger, "SMALLSERIAL"),
(None, postgresql.INTEGER, "SERIAL"),
(None, postgresql.BIGINT, "BIGSERIAL"),
(None, Integer().with_variant(BigInteger(), "postgresql"), "BIGSERIAL"),
(None, Integer().with_variant(postgresql.BIGINT, "postgresql"), "BIGSERIAL"),
((9, 2), Integer().with_variant(SmallInteger, "postgresql"), "SMALLSERIAL"),
(None, BITD(), "BIGSERIAL"),
]:
m = MetaData()
t = Table("t", m, Column("c", type_, primary_key=True))
if version:
dialect = postgresql.dialect()
dialect._get_server_version_info = Mock(return_value=version)
dialect.initialize(testing.db.connect())
else:
dialect = testing.db.dialect
ddl_compiler = dialect.ddl_compiler(dialect, schema.CreateTable(t))
eq_(ddl_compiler.get_column_specification(t.c.c), "c %s NOT NULL" % expected)
开发者ID:fuzzball81,项目名称:sqlalchemy,代码行数:35,代码来源:test_dialect.py
示例3: test_version_parsing
def test_version_parsing(self):
def mock_conn(res):
return Mock(
execute=Mock(return_value=Mock(scalar=Mock(return_value=res))))
dialect = postgresql.dialect()
for string, version in [
(
'PostgreSQL 8.3.8 on i686-redhat-linux-gnu, compiled by '
'GCC gcc (GCC) 4.1.2 20070925 (Red Hat 4.1.2-33)',
(8, 3, 8)),
(
'PostgreSQL 8.5devel on x86_64-unknown-linux-gnu, '
'compiled by GCC gcc (GCC) 4.4.2, 64-bit', (8, 5)),
(
'EnterpriseDB 9.1.2.2 on x86_64-unknown-linux-gnu, '
'compiled by gcc (GCC) 4.1.2 20080704 (Red Hat 4.1.2-50), '
'64-bit', (9, 1, 2)),
(
'[PostgreSQL 9.2.4 ] VMware vFabric Postgres 9.2.4.0 '
'release build 1080137', (9, 2, 4)),
(
'PostgreSQL 10devel on x86_64-pc-linux-gnu'
'compiled by gcc (GCC) 6.3.1 20170306, 64-bit', (10,)),
(
'PostgreSQL 10beta1 on x86_64-pc-linux-gnu, '
'compiled by gcc (GCC) 4.8.5 20150623 '
'(Red Hat 4.8.5-11), 64-bit', (10,))
]:
eq_(dialect._get_server_version_info(mock_conn(string)),
version)
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:32,代码来源:test_dialect.py
示例4: test_process_bind_param_str_for_postgres
def test_process_bind_param_str_for_postgres(self):
from sqlalchemy.dialects.postgresql.base import dialect
inet = self.INET()
output = inet.process_bind_param("127.1.2.3", dialect())
self.assertTrue(isinstance(output, str))
self.assertEqual(output, "127.1.2.3")
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:7,代码来源:test_inet.py
示例5: _raise_query
def _raise_query(self, q):
"""
useful for debugging. just do...
self._raise_query(q)
"""
c = q.compile(dialect=postgresql.dialect())
raise ValueError(c)
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:7,代码来源:test_compiler.py
示例6: test_functions
def test_functions(self):
from sqlalchemy.dialects.postgresql import base as postgresql
class MyUtcFunction(FunctionElement):
pass
@compiles(MyUtcFunction)
def visit_myfunc(element, compiler, **kw):
return "utcnow()"
@compiles(MyUtcFunction, 'postgresql')
def visit_myfunc(element, compiler, **kw):
return "timezone('utc', current_timestamp)"
self.assert_compile(
MyUtcFunction(),
"utcnow()",
use_default_dialect=True
)
self.assert_compile(
MyUtcFunction(),
"timezone('utc', current_timestamp)",
dialect=postgresql.dialect()
)
开发者ID:clones,项目名称:sqlalchemy,代码行数:24,代码来源:test_compiler.py
示例7: test_process_bind_param_other_dialect_with_data
def test_process_bind_param_other_dialect_with_data(self):
from sqlalchemy.dialects.sqlite.base import dialect
mapping = self.Mapping()
input = {u'foo': u'bâz'}
output = mapping.process_bind_param(input, dialect())
self.assertTrue(isinstance(output, str))
self.assertEqual(output, '{"foo": "b\\u00e2z"}')
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:7,代码来源:test_mapping.py
示例8: test_insert_returning
def test_insert_returning(self):
dialect = postgresql.dialect()
table1 = table('mytable',
column('myid', Integer),
column('name', String(128)),
column('description', String(128)),
)
i = insert(
table1,
values=dict(
name='foo')).returning(
table1.c.myid,
table1.c.name)
self.assert_compile(i,
'INSERT INTO mytable (name) VALUES '
'(%(name)s) RETURNING mytable.myid, '
'mytable.name', dialect=dialect)
i = insert(table1, values=dict(name='foo')).returning(table1)
self.assert_compile(i,
'INSERT INTO mytable (name) VALUES '
'(%(name)s) RETURNING mytable.myid, '
'mytable.name, mytable.description',
dialect=dialect)
i = insert(table1, values=dict(name='foo'
)).returning(func.length(table1.c.name))
self.assert_compile(i,
'INSERT INTO mytable (name) VALUES '
'(%(name)s) RETURNING length(mytable.name) '
'AS length_1', dialect=dialect)
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:30,代码来源:test_compiler.py
示例9: test_types
def test_types(self):
class MyType(TypeEngine):
pass
@compiles(MyType, 'sqlite')
def visit_type(type, compiler, **kw):
return "SQLITE_FOO"
@compiles(MyType, 'postgresql')
def visit_type(type, compiler, **kw):
return "POSTGRES_FOO"
from sqlalchemy.dialects.sqlite import base as sqlite
from sqlalchemy.dialects.postgresql import base as postgresql
self.assert_compile(
MyType(),
"SQLITE_FOO",
dialect=sqlite.dialect()
)
self.assert_compile(
MyType(),
"POSTGRES_FOO",
dialect=postgresql.dialect()
)
开发者ID:onetera,项目名称:scandatatransfer,代码行数:26,代码来源:test_compiler.py
示例10: test_serial_integer
def test_serial_integer(self):
for version, type_, expected in [
(None, Integer, 'SERIAL'),
(None, BigInteger, 'BIGSERIAL'),
((9, 1), SmallInteger, 'SMALLINT'),
((9, 2), SmallInteger, 'SMALLSERIAL'),
(None, postgresql.INTEGER, 'SERIAL'),
(None, postgresql.BIGINT, 'BIGSERIAL'),
]:
m = MetaData()
t = Table('t', m, Column('c', type_, primary_key=True))
if version:
dialect = postgresql.dialect()
dialect._get_server_version_info = Mock(return_value=version)
dialect.initialize(testing.db.connect())
else:
dialect = testing.db.dialect
ddl_compiler = dialect.ddl_compiler(dialect, schema.CreateTable(t))
eq_(
ddl_compiler.get_column_specification(t.c.c),
"c %s NOT NULL" % expected
)
开发者ID:EvaSDK,项目名称:sqlalchemy,代码行数:26,代码来源:test_dialect.py
示例11: test_process_bind_param_other_dialect_with_data
def test_process_bind_param_other_dialect_with_data(self):
from sqlalchemy.dialects.sqlite.base import dialect
array = Array(int)
input = [1, 2, 3]
output = array.process_bind_param(input, dialect())
self.assertTrue(isinstance(output, str))
self.assertEqual(output, '[1, 2, 3]')
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:7,代码来源:test_array.py
示例12: test_load_dialect_impl_postgresql
def test_load_dialect_impl_postgresql(self):
from sqlalchemy.dialects.postgresql.base import dialect
from sqlalchemy.dialects.postgresql import INET
inet = self.INET()
impl = inet.load_dialect_impl(dialect())
self.assertTrue(isinstance(impl, INET))
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:7,代码来源:test_inet.py
示例13: test_process_bind_param_str_for_sqlite
def test_process_bind_param_str_for_sqlite(self):
from sqlalchemy.dialects.sqlite.base import dialect
inet = self.INET()
output = inet.process_bind_param("10", dialect())
self.assertTrue(isinstance(output, str))
self.assertEqual(output, "10.0.0.0")
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:7,代码来源:test_inet.py
示例14: test_update_returning
def test_update_returning(self):
dialect = postgresql.dialect()
table1 = table(
'mytable',
column(
'myid', Integer),
column(
'name', String(128)),
column(
'description', String(128)))
u = update(
table1,
values=dict(
name='foo')).returning(
table1.c.myid,
table1.c.name)
self.assert_compile(u,
'UPDATE mytable SET name=%(name)s '
'RETURNING mytable.myid, mytable.name',
dialect=dialect)
u = update(table1, values=dict(name='foo')).returning(table1)
self.assert_compile(u,
'UPDATE mytable SET name=%(name)s '
'RETURNING mytable.myid, mytable.name, '
'mytable.description', dialect=dialect)
u = update(table1, values=dict(name='foo'
)).returning(func.length(table1.c.name))
self.assert_compile(
u,
'UPDATE mytable SET name=%(name)s '
'RETURNING length(mytable.name) AS length_1',
dialect=dialect)
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:32,代码来源:test_compiler.py
示例15: test_process_bind_param_uuid_for_sqlite
def test_process_bind_param_uuid_for_sqlite(self):
import uuid
from sqlalchemy.dialects.sqlite.base import dialect
guid = self.GUID()
input = uuid.uuid4()
output = guid.process_bind_param(input, dialect())
self.assertTrue(isinstance(output, str))
self.assertEqual(output, str(input))
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:8,代码来源:test_uuid.py
示例16: test_process_bind_param_ip_for_sqlite
def test_process_bind_param_ip_for_sqlite(self):
import IPy
from sqlalchemy.dialects.sqlite.base import dialect
inet = self.INET()
input = IPy.IP("::1")
output = inet.process_bind_param(input, dialect())
self.assertTrue(isinstance(output, str))
self.assertEqual(output, "::1")
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:9,代码来源:test_inet.py
示例17: test_process_bind_param_ip_for_postgres
def test_process_bind_param_ip_for_postgres(self):
import IPy
from sqlalchemy.dialects.postgresql.base import dialect
inet = self.INET()
input = IPy.IP("127.0.0.1")
output = inet.process_bind_param(input, dialect())
self.assertTrue(isinstance(output, str))
self.assertEqual(output, str(input))
开发者ID:curvetips,项目名称:s4u.sqlalchemy,代码行数:9,代码来源:test_inet.py
示例18: test_exclude_constraint_min
def test_exclude_constraint_min(self):
m = MetaData()
tbl = Table('testtbl', m,
Column('room', Integer, primary_key=True))
cons = ExcludeConstraint(('room', '='))
tbl.append_constraint(cons)
self.assert_compile(schema.AddConstraint(cons),
'ALTER TABLE testtbl ADD EXCLUDE USING gist '
'(room WITH =)',
dialect=postgresql.dialect())
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:10,代码来源:test_compiler.py
示例19: test_create_index_with_using
def test_create_index_with_using(self):
m = MetaData()
tbl = Table('testtbl', m, Column('data', String))
idx1 = Index('test_idx1', tbl.c.data)
idx2 = Index('test_idx2', tbl.c.data, postgresql_using='btree')
idx3 = Index('test_idx3', tbl.c.data, postgresql_using='hash')
self.assert_compile(schema.CreateIndex(idx1),
'CREATE INDEX test_idx1 ON testtbl '
'(data)',
dialect=postgresql.dialect())
self.assert_compile(schema.CreateIndex(idx2),
'CREATE INDEX test_idx2 ON testtbl '
'USING btree (data)',
dialect=postgresql.dialect())
self.assert_compile(schema.CreateIndex(idx3),
'CREATE INDEX test_idx3 ON testtbl '
'USING hash (data)',
dialect=postgresql.dialect())
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:20,代码来源:test_compiler.py
示例20: test_create_index_with_ops
def test_create_index_with_ops(self):
m = MetaData()
tbl = Table('testtbl', m,
Column('data', String),
Column('data2', Integer, key='d2'))
idx = Index('test_idx1', tbl.c.data,
postgresql_ops={'data': 'text_pattern_ops'})
idx2 = Index('test_idx2', tbl.c.data, tbl.c.d2,
postgresql_ops={'data': 'text_pattern_ops',
'd2': 'int4_ops'})
self.assert_compile(schema.CreateIndex(idx),
'CREATE INDEX test_idx1 ON testtbl '
'(data text_pattern_ops)',
dialect=postgresql.dialect())
self.assert_compile(schema.CreateIndex(idx2),
'CREATE INDEX test_idx2 ON testtbl '
'(data text_pattern_ops, data2 int4_ops)',
dialect=postgresql.dialect())
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:21,代码来源:test_compiler.py
注:本文中的sqlalchemy.dialects.postgresql.base.dialect函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论