本文整理汇总了Python中query_fixture.query_fixture函数的典型用法代码示例。如果您正苦于以下问题:Python query_fixture函数的具体用法?Python query_fixture怎么用?Python query_fixture使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了query_fixture函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_select_single_row_bool_result
def test_select_single_row_bool_result(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'SELECT TRUE') as query:
cursor.execute(query)
row = cursor.fetchone()
assert row == [True]
with query_fixture(cursor, configuration, 'SELECT FALSE') as query:
cursor.execute(query)
row = cursor.fetchone()
assert row == [False]
开发者ID:rlugojr,项目名称:turbodbc,代码行数:10,代码来源:test_cursor_select.py
示例2: test_fetchmany_with_default_arraysize
def test_fetchmany_with_default_arraysize(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'SELECT MULTIPLE INTEGERS') as query:
cursor.execute(query)
rows = cursor.fetchmany()
assert len(rows) == 1
assert rows[0] == [42]
开发者ID:rlugojr,项目名称:turbodbc,代码行数:7,代码来源:test_cursor_select.py
示例3: test_arrow_string_column
def test_arrow_string_column(dsn, configuration, strings_as_dictionary):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT UNICODE') as table_name:
cursor.execute('INSERT INTO {} VALUES (?)'.format(table_name), [u'unicode \u2665'])
cursor.execute('SELECT a FROM {}'.format(table_name))
result = cursor.fetchallarrow(strings_as_dictionary=strings_as_dictionary)
assert result.column(0).to_pylist() == [u'unicode \u2665']
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:7,代码来源:test_select_arrow.py
示例4: test_arrow_reference_count
def test_arrow_reference_count(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
cursor.execute("SELECT a FROM {}".format(table_name))
result = cursor.fetchallarrow()
gc.collect()
assert sys.getrefcount(result) == 2
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:7,代码来源:test_select_arrow.py
示例5: test_numpy_double_column
def test_numpy_double_column(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'SELECT DOUBLE') as query:
cursor.execute(query)
results = cursor.fetchallnumpy()
expected = MaskedArray([3.14], mask=[0])
assert results[_fix_case(configuration, 'a')].dtype == numpy.float64
assert_equal(results[_fix_case(configuration, 'a')], expected)
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_select_numpy.py
示例6: test_numpy_empty_column
def test_numpy_empty_column(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
cursor.execute("SELECT a FROM {}".format(table_name))
results = cursor.fetchallnumpy()
assert isinstance(results, OrderedDict)
assert len(results) == 1 # ncols
assert isinstance(results[_fix_case(configuration, 'a')], MaskedArray)
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_select_numpy.py
示例7: test_numpy_string_column_with_null
def test_numpy_string_column_with_null(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT STRING') as table_name:
cursor.execute('INSERT INTO {} VALUES (?)'.format(table_name), [None])
cursor.execute('SELECT a FROM {}'.format(table_name))
results = cursor.fetchallnumpy()
expected = MaskedArray([None], mask=[0], dtype=numpy.object_)
assert_equal(results[_fix_case(configuration, 'a')], expected)
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_select_numpy.py
示例8: test_insert_duplicate_uniquecol_raises
def test_insert_duplicate_uniquecol_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT DUPLICATE UNIQUECOL') as table_name:
with pytest.raises(DatabaseError) as ex:
cursor.execute("INSERT INTO {table_name} VALUES (1)".format(table_name=table_name))
# some databases (e.g. exasol) report failure not in the execute statement above, but only
# when closing the odbc handle, i.e. at cursor.close:
cursor.close()
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_cursor_basics.py
示例9: test_rowcount_is_reset_after_execute_raises
def test_rowcount_is_reset_after_execute_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
cursor.execute("INSERT INTO {} VALUES (?)".format(table_name), [42])
assert cursor.rowcount == 1
with pytest.raises(Error):
cursor.execute("this is not even a valid SQL statement")
assert cursor.rowcount == -1
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_cursor_basics.py
示例10: test_arrow_empty_column
def test_arrow_empty_column(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
cursor.execute("SELECT a FROM {}".format(table_name))
result = cursor.fetchallarrow()
assert isinstance(result, pa.Table)
assert result.num_columns == 1
assert result.num_rows == 0
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_select_arrow.py
示例11: _test_insert_one
def _test_insert_one(configuration, fixture_name, data):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, fixture_name) as table_name:
cursor.execute("INSERT INTO {} VALUES (?)".format(table_name), data)
assert 1 == cursor.rowcount
cursor.execute("SELECT a FROM {}".format(table_name))
inserted = cursor.fetchall()
assert [list(data)] == inserted
开发者ID:rlugojr,项目名称:turbodbc,代码行数:8,代码来源:test_cursor_insert.py
示例12: test_arrow_two_columns
def test_arrow_two_columns(dsn, configuration):
with open_cursor(configuration, rows_to_buffer=1) as cursor:
with query_fixture(cursor, configuration, 'INSERT TWO INTEGER COLUMNS') as table_name:
cursor.executemany("INSERT INTO {} VALUES (?, ?)".format(table_name),
[[1, 42], [2, 41]])
cursor.execute("SELECT a, b FROM {} ORDER BY a".format(table_name))
result = list(cursor.fetcharrowbatches())
assert len(result) == 2
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_select_arrow.py
示例13: test_insert_no_parameter_list
def test_insert_no_parameter_list(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
cursor.executemany("INSERT INTO {} VALUES (?)".format(table_name))
assert 0 == cursor.rowcount
cursor.execute("SELECT a FROM {}".format(table_name))
inserted = [list(row) for row in cursor.fetchall()]
assert 0 == len(inserted)
开发者ID:rlugojr,项目名称:turbodbc,代码行数:8,代码来源:test_cursor_insert.py
示例14: test_arrow_string_column_with_null
def test_arrow_string_column_with_null(dsn, configuration, strings_as_dictionary):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT STRING') as table_name:
cursor.execute('INSERT INTO {} VALUES (?)'.format(table_name), [None])
cursor.execute('SELECT a FROM {}'.format(table_name))
result = cursor.fetchallarrow(strings_as_dictionary=strings_as_dictionary)
result.column(0).null_count == 1
result.column(0).to_pylist() == [None]
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:8,代码来源:test_select_arrow.py
示例15: test_fetchmany_with_bad_arraysize_parameter_raises
def test_fetchmany_with_bad_arraysize_parameter_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'SELECT MULTIPLE INTEGERS') as query:
cursor.execute(query)
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchmany(-1)
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchmany(0)
开发者ID:rlugojr,项目名称:turbodbc,代码行数:9,代码来源:test_cursor_select.py
示例16: test_numpy_unicode_column
def test_numpy_unicode_column(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT UNICODE') as table_name:
cursor.execute('INSERT INTO {} VALUES (?)'.format(table_name), [u'unicode \u2665'])
cursor.execute('SELECT a FROM {}'.format(table_name))
results = cursor.fetchallnumpy()
expected = MaskedArray([u'unicode \u2665'], mask=[0], dtype=numpy.object_)
assert results[_fix_case(configuration, 'a')].dtype == numpy.object_
assert_equal(results[_fix_case(configuration, 'a')], expected)
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:9,代码来源:test_select_numpy.py
示例17: test_arrow_timelike_column_with_null
def test_arrow_timelike_column_with_null(dsn, configuration):
fill_value = 0;
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT TIMESTAMP') as table_name:
cursor.execute('INSERT INTO {} VALUES (?)'.format(table_name), [None])
cursor.execute('SELECT a FROM {}'.format(table_name))
result = cursor.fetchallarrow()
assert result.column(0).to_pylist() == [None]
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:9,代码来源:test_select_arrow.py
示例18: test_arrow_date_column
def test_arrow_date_column(dsn, configuration):
date = datetime.date(2015, 12, 31)
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT DATE') as table_name:
cursor.execute('INSERT INTO {} VALUES (?)'.format(table_name), [date])
cursor.execute('SELECT a FROM {}'.format(table_name))
result = cursor.fetchallarrow()
result.column(0).to_pylist() == [datetime.date(2015, 12, 31)]
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:9,代码来源:test_select_arrow.py
示例19: test_numpy_binary_column_larger_than_batch_size
def test_numpy_binary_column_larger_than_batch_size(dsn, configuration):
with open_cursor(configuration, rows_to_buffer=2) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
cursor.executemany("INSERT INTO {} VALUES (?)".format(table_name),
[[1], [2], [3], [4], [5]])
cursor.execute("SELECT a FROM {} ORDER BY a".format(table_name))
results = cursor.fetchallnumpy()
expected = MaskedArray([1, 2, 3, 4, 5], mask=False)
assert_equal(results[_fix_case(configuration, 'a')], expected)
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:9,代码来源:test_select_numpy.py
示例20: test_arrow_binary_column_larger_than_batch_size
def test_arrow_binary_column_larger_than_batch_size(dsn, configuration):
with open_cursor(configuration, rows_to_buffer=2) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
cursor.executemany("INSERT INTO {} VALUES (?)".format(table_name),
[[1], [2], [3], [4], [5]])
cursor.execute("SELECT a FROM {} ORDER BY a".format(table_name))
result = cursor.fetchallarrow()
assert isinstance(result, pa.Table)
assert result.column(0).to_pylist() == [1, 2, 3, 4, 5]
开发者ID:blue-yonder,项目名称:turbodbc,代码行数:9,代码来源:test_select_arrow.py
注:本文中的query_fixture.query_fixture函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论