本文整理汇总了Python中sqlalchemy.testing.against函数的典型用法代码示例。如果您正苦于以下问题:Python against函数的具体用法?Python against怎么用?Python against使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了against函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _non_default_isolation_level
def _non_default_isolation_level(self):
if testing.against('sqlite'):
return 'READ UNCOMMITTED'
elif testing.against('postgresql'):
return 'SERIALIZABLE'
elif testing.against('mysql'):
return "SERIALIZABLE"
else:
assert False, "non default isolation level not known"
开发者ID:SvenDowideit,项目名称:clearlinux,代码行数:9,代码来源:test_transaction.py
示例2: test_set
def test_set(self):
with testing.expect_deprecated('Manually quoting SET value literals'):
e1, e2 = mysql.SET("'a'", "'b'"), mysql.SET("'a'", "'b'")
e4 = mysql.SET("'a'", "b")
e5 = mysql.SET("'a'", "'b'", quoting="quoted")
set_table = Table('mysql_set', self.metadata,
Column('e1', e1),
Column('e2', e2, nullable=False),
Column('e3', mysql.SET("a", "b")),
Column('e4', e4),
Column('e5', e5)
)
eq_(colspec(set_table.c.e1),
"e1 SET('a','b')")
eq_(colspec(set_table.c.e2),
"e2 SET('a','b') NOT NULL")
eq_(colspec(set_table.c.e3),
"e3 SET('a','b')")
eq_(colspec(set_table.c.e4),
"e4 SET('''a''','b')")
eq_(colspec(set_table.c.e5),
"e5 SET('a','b')")
set_table.create()
assert_raises(exc.DBAPIError, set_table.insert().execute,
e1=None, e2=None, e3=None, e4=None)
if testing.against("+oursql"):
assert_raises(exc.StatementError, set_table.insert().execute,
e1='c', e2='c', e3='c', e4='c')
set_table.insert().execute(e1='a', e2='a', e3='a', e4="'a'", e5="a,b")
set_table.insert().execute(e1='b', e2='b', e3='b', e4='b', e5="a,b")
res = set_table.select().execute().fetchall()
if not testing.against("+oursql"):
# oursql receives this for first row:
# (set(['']), set(['']), set(['']), set(['']), None),
# but based on ...OS? MySQL version? not clear.
# not worth testing.
expected = []
expected.extend([
(set(['a']), set(['a']), set(['a']), set(["'a'"]), set(['a', 'b'])),
(set(['b']), set(['b']), set(['b']), set(['b']), set(['a', 'b']))
])
eq_(res, expected)
开发者ID:SmartTeleMax,项目名称:sqlalchemy,代码行数:52,代码来源:test_types.py
示例3: define_tables
def define_tables(cls, metadata):
if testing.against('oracle'):
fk_args = dict(deferrable=True, initially='deferred')
elif testing.against('mysql'):
fk_args = {}
else:
fk_args = dict(onupdate='cascade')
Table('users', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True))
Table('addresses', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('user_id', Integer, ForeignKey('users.id', **fk_args)))
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:15,代码来源:test_hasparent.py
示例4: define_tables
def define_tables(cls, metadata):
if testing.against("oracle"):
fk_args = dict(deferrable=True, initially="deferred")
elif testing.against("mysql"):
fk_args = {}
else:
fk_args = dict(onupdate="cascade")
Table("users", metadata, Column("id", Integer, primary_key=True, test_needs_autoincrement=True))
Table(
"addresses",
metadata,
Column("id", Integer, primary_key=True, test_needs_autoincrement=True),
Column("user_id", Integer, ForeignKey("users.id", **fk_args)),
)
开发者ID:GitHublong,项目名称:sqlalchemy,代码行数:15,代码来源:test_hasparent.py
示例5: test_set
def test_set(self):
with testing.expect_deprecated("Manually quoting SET value literals"):
e1, e2 = mysql.SET("'a'", "'b'"), mysql.SET("'a'", "'b'")
set_table = Table(
"mysql_set",
self.metadata,
Column("e1", e1),
Column("e2", e2, nullable=False),
Column("e3", mysql.SET("a", "b")),
Column("e4", mysql.SET("'a'", "b")),
Column("e5", mysql.SET("'a'", "'b'", quoting="quoted")),
)
eq_(colspec(set_table.c.e1), "e1 SET('a','b')")
eq_(colspec(set_table.c.e2), "e2 SET('a','b') NOT NULL")
eq_(colspec(set_table.c.e3), "e3 SET('a','b')")
eq_(colspec(set_table.c.e4), "e4 SET('''a''','b')")
eq_(colspec(set_table.c.e5), "e5 SET('a','b')")
set_table.create()
assert_raises(exc.DBAPIError, set_table.insert().execute, e1=None, e2=None, e3=None, e4=None)
if testing.against("+oursql"):
assert_raises(exc.StatementError, set_table.insert().execute, e1="c", e2="c", e3="c", e4="c")
set_table.insert().execute(e1="a", e2="a", e3="a", e4="'a'", e5="a,b")
set_table.insert().execute(e1="b", e2="b", e3="b", e4="b", e5="a,b")
res = set_table.select().execute().fetchall()
if testing.against("+oursql"):
expected = [
# 1st row with all c's, data truncated
(set([""]), set([""]), set([""]), set([""]), None)
]
else:
expected = []
expected.extend(
[
(set(["a"]), set(["a"]), set(["a"]), set(["'a'"]), set(["a", "b"])),
(set(["b"]), set(["b"]), set(["b"]), set(["b"]), set(["a", "b"])),
]
)
eq_(res, expected)
开发者ID:Julian,项目名称:sqlalchemy,代码行数:48,代码来源:test_types.py
示例6: test_column_accessor_sqlite_raw
def test_column_accessor_sqlite_raw(self):
users = self.tables.users
users.insert().execute(
dict(user_id=1, user_name='john'),
)
r = text(
"select users.user_id, users.user_name "
"from users "
"UNION select users.user_id, "
"users.user_name from users",
bind=testing.db).execution_options(sqlite_raw_colnames=True). \
execute().first()
if testing.against("sqlite < 3.10.0"):
not_in_('user_id', r)
not_in_('user_name', r)
eq_(r['users.user_id'], 1)
eq_(r['users.user_name'], "john")
eq_(list(r.keys()), ["users.user_id", "users.user_name"])
else:
not_in_('users.user_id', r)
not_in_('users.user_name', r)
eq_(r['user_id'], 1)
eq_(r['user_name'], "john")
eq_(list(r.keys()), ["user_id", "user_name"])
开发者ID:mattastica,项目名称:sqlalchemy,代码行数:29,代码来源:test_resultset.py
示例7: test_limit
def test_limit(self):
"""test limit operations combined with lazy-load relationships."""
users, items, order_items, orders, Item, \
User, Address, Order, addresses = (
self.tables.users,
self.tables.items,
self.tables.order_items,
self.tables.orders,
self.classes.Item,
self.classes.User,
self.classes.Address,
self.classes.Order,
self.tables.addresses)
mapper(Item, items)
mapper(Order, orders, properties={
'items': relationship(Item, secondary=order_items, lazy='select')
})
mapper(User, users, properties={
'addresses': relationship(
mapper(Address, addresses), lazy='select'),
'orders': relationship(Order, lazy='select')
})
sess = create_session()
q = sess.query(User)
if testing.against('mssql'):
result = q.limit(2).all()
assert self.static.user_all_result[:2] == result
else:
result = q.limit(2).offset(1).all()
assert self.static.user_all_result[1:3] == result
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:34,代码来源:test_lazy_relations.py
示例8: test_reflect_nvarchar
def test_reflect_nvarchar(self):
metadata = self.metadata
Table(
"tnv",
metadata,
Column("nv_data", sqltypes.NVARCHAR(255)),
Column("c_data", sqltypes.NCHAR(20)),
)
metadata.create_all()
m2 = MetaData(testing.db)
t2 = Table("tnv", m2, autoload=True)
assert isinstance(t2.c.nv_data.type, sqltypes.NVARCHAR)
assert isinstance(t2.c.c_data.type, sqltypes.NCHAR)
if testing.against("oracle+cx_oracle"):
assert isinstance(
t2.c.nv_data.type.dialect_impl(testing.db.dialect),
cx_oracle._OracleUnicodeStringNCHAR,
)
assert isinstance(
t2.c.c_data.type.dialect_impl(testing.db.dialect),
cx_oracle._OracleUnicodeStringNCHAR,
)
data = u("m’a réveillé.")
with testing.db.connect() as conn:
conn.execute(t2.insert(), dict(nv_data=data, c_data=data))
nv_data, c_data = conn.execute(t2.select()).first()
eq_(nv_data, data)
eq_(c_data, data + (" " * 7)) # char is space padded
assert isinstance(nv_data, util.text_type)
assert isinstance(c_data, util.text_type)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:33,代码来源:test_types.py
示例9: test_text_doesnt_explode
def test_text_doesnt_explode(self):
for s in [
select(
[
case(
[
(
info_table.c.info == 'pk_4_data',
text("'yes'"))],
else_=text("'no'"))
]).order_by(info_table.c.info),
select(
[
case(
[
(
info_table.c.info == 'pk_4_data',
literal_column("'yes'"))],
else_=literal_column("'no'")
)]
).order_by(info_table.c.info),
]:
if testing.against("firebird"):
eq_(s.execute().fetchall(), [
('no ', ), ('no ', ), ('no ', ), ('yes', ),
('no ', ), ('no ', ),
])
else:
eq_(s.execute().fetchall(), [
('no', ), ('no', ), ('no', ), ('yes', ),
('no', ), ('no', ),
])
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:35,代码来源:test_case_statement.py
示例10: test_int_default_none_on_insert_reflected
def test_int_default_none_on_insert_reflected(self):
metadata = self.metadata
Table('x', metadata,
Column('y', Integer,
server_default='5', primary_key=True),
Column('data', String(10)),
implicit_returning=False
)
metadata.create_all()
m2 = MetaData(metadata.bind)
t2 = Table('x', m2, autoload=True, implicit_returning=False)
r = t2.insert().execute(data='data')
eq_(r.inserted_primary_key, [None])
if testing.against('sqlite'):
eq_(
t2.select().execute().fetchall(),
[(1, 'data')]
)
else:
eq_(
t2.select().execute().fetchall(),
[(5, 'data')]
)
开发者ID:Affirm,项目名称:sqlalchemy,代码行数:25,代码来源:test_defaults.py
示例11: test_updatemany
def test_updatemany(self):
# MySQL-Python 1.2.2 breaks functions in execute_many :(
if (testing.against('mysql+mysqldb') and
testing.db.dialect.dbapi.version_info[:3] == (1, 2, 2)):
return
t.insert().execute({}, {}, {})
t.update(t.c.col1 == sa.bindparam('pkval')).execute(
{'pkval': 51, 'col7': None, 'col8': None, 'boolcol1': False})
t.update(t.c.col1 == sa.bindparam('pkval')).execute(
{'pkval': 51},
{'pkval': 52},
{'pkval': 53})
l = t.select().execute()
ctexec = currenttime.scalar()
today = datetime.date.today()
eq_(l.fetchall(),
[(51, 'im the update', f2, ts, ts, ctexec, False, False,
13, today, 'py', 'hi'),
(52, 'im the update', f2, ts, ts, ctexec, True, False,
13, today, 'py', 'hi'),
(53, 'im the update', f2, ts, ts, ctexec, True, False,
13, today, 'py', 'hi')])
开发者ID:Affirm,项目名称:sqlalchemy,代码行数:26,代码来源:test_defaults.py
示例12: define_temp_tables
def define_temp_tables(cls, metadata):
# cheat a bit, we should fix this with some dialect-level
# temp table fixture
if testing.against("oracle"):
kw = {
'prefixes': ["GLOBAL TEMPORARY"],
'oracle_on_commit': 'PRESERVE ROWS'
}
else:
kw = {
'prefixes': ["TEMPORARY"],
}
user_tmp = Table(
"user_tmp", metadata,
Column("id", sa.INT, primary_key=True),
Column('name', sa.VARCHAR(50)),
Column('foo', sa.INT),
sa.UniqueConstraint('name', name='user_tmp_uq'),
sa.Index("user_tmp_ix", "foo"),
**kw
)
if testing.requires.view_reflection.enabled and \
testing.requires.temporary_views.enabled:
event.listen(
user_tmp, "after_create",
DDL("create temporary view user_tmp_v as "
"select * from user_tmp")
)
event.listen(
user_tmp, "before_drop",
DDL("drop view user_tmp_v")
)
开发者ID:DanielPearl,项目名称:top_commodities,代码行数:33,代码来源:test_reflection.py
示例13: _test_lastrow_accessor
def _test_lastrow_accessor(self, table_, values, assertvalues):
"""Tests the inserted_primary_key and lastrow_has_id() functions."""
def insert_values(engine, table_, values):
"""
Inserts a row into a table, returns the full list of values
INSERTed including defaults that fired off on the DB side and
detects rows that had defaults and post-fetches.
"""
# verify implicit_returning is working
if engine.dialect.implicit_returning:
ins = table_.insert()
comp = ins.compile(engine, column_keys=list(values))
if not set(values).issuperset(
c.key for c in table_.primary_key
):
is_(bool(comp.returning), True)
result = engine.execute(table_.insert(), **values)
ret = values.copy()
for col, id_ in zip(
table_.primary_key, result.inserted_primary_key
):
ret[col.key] = id_
if result.lastrow_has_defaults():
criterion = and_(
*[
col == id_
for col, id_ in zip(
table_.primary_key, result.inserted_primary_key
)
]
)
row = engine.execute(table_.select(criterion)).first()
for c in table_.c:
ret[c.key] = row[c]
return ret
if testing.against("firebird", "postgresql", "oracle", "mssql"):
assert testing.db.dialect.implicit_returning
if testing.db.dialect.implicit_returning:
test_engines = [
engines.testing_engine(options={"implicit_returning": False}),
engines.testing_engine(options={"implicit_returning": True}),
]
else:
test_engines = [testing.db]
for engine in test_engines:
try:
table_.create(bind=engine, checkfirst=True)
i = insert_values(engine, table_, values)
eq_(i, assertvalues)
finally:
table_.drop(bind=engine)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:59,代码来源:test_insert_exec.py
示例14: test_outer_joinedload_w_limit
def test_outer_joinedload_w_limit(self):
User = self.classes.User
sess = Session()
q = sess.query(User).options(
joinedload(User.addresses, innerjoin=False)
)
if testing.against("postgresql"):
q = q.with_for_update(of=User)
else:
q = q.with_for_update()
q = q.limit(1)
if testing.against("oracle"):
assert_raises_message(exc.DatabaseError, "ORA-02014", q.all)
else:
q.all()
sess.close()
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:19,代码来源:test_lockmode.py
示例15: get_objects_skipping_sqlite_issue
def get_objects_skipping_sqlite_issue():
# pysqlite keeps adding weakref objects which only
# get reset after 220 iterations. We'd like to keep these
# tests under 50 iterations and ideally about ten, so
# just filter them out so that we get a "flatline" more quickly.
if testing.against("sqlite+pysqlite"):
return [o for o in gc.get_objects()
if not isinstance(o, weakref.ref)]
else:
return gc.get_objects()
开发者ID:aruns807,项目名称:sqlalchemy,代码行数:10,代码来源:test_memusage.py
示例16: setup_class
def setup_class(cls):
global metadata, t1, t2, t3
metadata = MetaData(testing.db)
t1 = Table(u('unitable1'), metadata,
Column(u('méil'), Integer, primary_key=True),
Column(ue('\u6e2c\u8a66'), Integer),
test_needs_fk=True,
)
t2 = Table(
u('Unitéble2'),
metadata,
Column(
u('méil'),
Integer,
primary_key=True,
key="a"),
Column(
ue('\u6e2c\u8a66'),
Integer,
ForeignKey(
u('unitable1.méil')),
key="b"),
test_needs_fk=True,
)
# Few DBs support Unicode foreign keys
if testing.against('sqlite'):
t3 = Table(ue('\u6e2c\u8a66'), metadata,
Column(ue('\u6e2c\u8a66_id'), Integer, primary_key=True,
autoincrement=False),
Column(ue('unitable1_\u6e2c\u8a66'), Integer,
ForeignKey(ue('unitable1.\u6e2c\u8a66'))
),
Column(u('Unitéble2_b'), Integer,
ForeignKey(u('Unitéble2.b'))
),
Column(ue('\u6e2c\u8a66_self'), Integer,
ForeignKey(ue('\u6e2c\u8a66.\u6e2c\u8a66_id'))
),
test_needs_fk=True,
)
else:
t3 = Table(ue('\u6e2c\u8a66'), metadata,
Column(ue('\u6e2c\u8a66_id'), Integer, primary_key=True,
autoincrement=False),
Column(ue('unitable1_\u6e2c\u8a66'), Integer),
Column(u('Unitéble2_b'), Integer),
Column(ue('\u6e2c\u8a66_self'), Integer),
test_needs_fk=True,
)
metadata.create_all()
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:52,代码来源:test_unicode.py
示例17: _test_get_columns
def _test_get_columns(self, schema=None, table_type="table"):
meta = MetaData(testing.db)
users, addresses, dingalings = self.tables.users, self.tables.email_addresses, self.tables.dingalings
table_names = ["users", "email_addresses"]
if table_type == "view":
table_names = ["users_v", "email_addresses_v"]
insp = inspect(meta.bind)
for table_name, table in zip(table_names, (users, addresses)):
schema_name = schema
cols = insp.get_columns(table_name, schema=schema_name)
self.assert_(len(cols) > 0, len(cols))
# should be in order
for i, col in enumerate(table.columns):
eq_(col.name, cols[i]["name"])
ctype = cols[i]["type"].__class__
ctype_def = col.type
if isinstance(ctype_def, sa.types.TypeEngine):
ctype_def = ctype_def.__class__
# Oracle returns Date for DateTime.
if testing.against("oracle") and ctype_def in (sql_types.Date, sql_types.DateTime):
ctype_def = sql_types.Date
# assert that the desired type and return type share
# a base within one of the generic types.
self.assert_(
len(
set(ctype.__mro__)
.intersection(ctype_def.__mro__)
.intersection(
[
sql_types.Integer,
sql_types.Numeric,
sql_types.DateTime,
sql_types.Date,
sql_types.Time,
sql_types.String,
sql_types._Binary,
]
)
)
> 0,
"%s(%s), %s(%s)" % (col.name, col.type, cols[i]["name"], ctype),
)
if not col.primary_key:
assert cols[i]["default"] is None
开发者ID:nakagami,项目名称:sqlalchemy,代码行数:51,代码来源:test_reflection.py
示例18: test_outer_joinedload_wo_limit
def test_outer_joinedload_wo_limit(self):
User = self.classes.User
sess = Session()
q = sess.query(User).options(
joinedload(User.addresses, innerjoin=False)
)
if testing.against("postgresql"):
q = q.with_for_update(of=User)
else:
q = q.with_for_update()
q.all()
sess.close()
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:14,代码来源:test_lockmode.py
示例19: test_ilike
def test_ilike(self):
users.insert().execute(
{"user_id": 1, "user_name": "one"},
{"user_id": 2, "user_name": "TwO"},
{"user_id": 3, "user_name": "ONE"},
{"user_id": 4, "user_name": "OnE"},
)
eq_(select([users.c.user_id]).where(users.c.user_name.ilike("one")).execute().fetchall(), [(1,), (3,), (4,)])
eq_(select([users.c.user_id]).where(users.c.user_name.ilike("TWO")).execute().fetchall(), [(2,)])
if testing.against("postgresql"):
eq_(select([users.c.user_id]).where(users.c.user_name.like("one")).execute().fetchall(), [(1,)])
eq_(select([users.c.user_id]).where(users.c.user_name.like("TWO")).execute().fetchall(), [])
开发者ID:t3573393,项目名称:sqlalchemy,代码行数:15,代码来源:test_query.py
示例20: test_int_default_none_on_insert
def test_int_default_none_on_insert(self):
metadata = self.metadata
t = Table(
"x",
metadata,
Column("y", Integer, server_default="5", primary_key=True),
Column("data", String(10)),
implicit_returning=False,
)
assert t._autoincrement_column is None
metadata.create_all()
r = t.insert().execute(data="data")
eq_(r.inserted_primary_key, [None])
if testing.against("sqlite"):
eq_(t.select().execute().fetchall(), [(1, "data")])
else:
eq_(t.select().execute().fetchall(), [(5, "data")])
开发者ID:duesenfranz,项目名称:sqlalchemy,代码行数:17,代码来源:test_defaults.py
注:本文中的sqlalchemy.testing.against函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论