Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
TColumnType(col_name='qty',
col_type=TTypeInfo(type=1, encoding=0, nullable=True,
is_array=False, precision=0,
scale=0, comp_param=0),
is_reserved_keyword=False, src_name=''),
TColumnType(col_name='price',
col_type=TTypeInfo(type=3, encoding=0, nullable=True,
is_array=False, precision=0,
scale=0, comp_param=0),
is_reserved_keyword=False, src_name=''),
TColumnType(col_name='vol',
col_type=TTypeInfo(type=3, encoding=0, nullable=True,
is_array=False, precision=0,
scale=0, comp_param=0),
is_reserved_keyword=False, src_name='')]
result = _extract_column_details(data)
expected = [
ColumnDetails(name='date_', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT',
is_array=False),
ColumnDetails(name='trans', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT',
is_array=False),
ColumnDetails(name='symbol', type='STR', nullable=True,
precision=0, scale=0, comp_param=32,
encoding='DICT',
is_array=False),
ColumnDetails(name='qty', type='INT', nullable=True, precision=0,
scale=0, comp_param=0, encoding='NONE',
is_array=False),
ColumnDetails(name='price', type='FLOAT', nullable=True,
def _get_schema_using_query(self, query):
with self._execute(query, results=True) as result:
# resets the state of the cursor and closes operation
result.cursor.fetchall()
names, ibis_types = self._adapt_types(
_extract_column_details(result.cursor._result.row_set.row_desc)
)
return sch.Schema(names, ibis_types)
Returns
-------
details: List[tuples]
Examples
--------
>>> con.get_table_details('stocks')
[ColumnDetails(name='date_', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT'),
ColumnDetails(name='trans', type='STR', nullable=True, precision=0,
scale=0, comp_param=32, encoding='DICT'),
...
]
"""
details = self._client.get_table_details(self._session, table_name)
return _extract_column_details(details.row_desc)