Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_type_info(self, curcol):
""" Reads TYPE_INFO structure (http://msdn.microsoft.com/en-us/library/dd358284.aspx)
:param curcol: An instance of :class:`Column` that will receive read information
"""
r = self._reader
# User defined data type of the column
curcol.column_usertype = r.get_uint() if IS_TDS72_PLUS(self) else r.get_usmallint()
curcol.flags = r.get_usmallint() # Flags
curcol.column_nullable = curcol.flags & Column.fNullable
curcol.column_writeable = (curcol.flags & Column.fReadWrite) > 0
curcol.column_identity = (curcol.flags & Column.fIdentity) > 0
type_id = r.get_byte()
type_class = self._tds._type_map.get(type_id)
if not type_class:
raise InterfaceError('Invalid type id', type_id)
curcol.type = type_class.from_stream(r)
def get_type_info(self, curcol):
""" Reads TYPE_INFO structure (http://msdn.microsoft.com/en-us/library/dd358284.aspx)
:param curcol: An instance of :class:`Column` that will receive read information
"""
r = self._reader
# User defined data type of the column
curcol.column_usertype = r.get_uint() if IS_TDS72_PLUS(self) else r.get_usmallint()
curcol.flags = r.get_usmallint() # Flags
curcol.column_nullable = curcol.flags & Column.fNullable
curcol.column_writeable = (curcol.flags & Column.fReadWrite) > 0
curcol.column_identity = (curcol.flags & Column.fIdentity) > 0
type_id = r.get_byte()
type_class = self._tds._type_map.get(type_id)
if not type_class:
raise InterfaceError('Invalid type id', type_id)
curcol.type = type_class.from_stream(r)
Value can also be of a special types:
- An instance of :class:`Column`, in which case it is just returned.
- An instance of :class:`output`, in which case parameter will become
an output parameter.
- A singleton :var:`default`, in which case default value will be passed
into a stored proc.
:param name: Name of the parameter, will populate column_name property of returned column.
:param value: Value of the parameter, also used to guess the type of parameter.
:return: An instance of :class:`Column`
"""
if isinstance(value, Column):
value.column_name = name
return value
column = Column()
column.column_name = name
column.flags = 0
if isinstance(value, output):
column.flags |= fByRefValue
value = value.value
if value is default:
column.flags = fDefaultValue
value = None
column.value = value
if value is None:
column.type = self.conn.NVarChar(1, collation=self.conn.collation)
elif isinstance(value, bool):
column.type = BitN()
elif isinstance(value, six.integer_types):
if -2 ** 31 <= value <= 2 ** 31 - 1:
column.type = IntN(4)
def process_param(self):
""" Reads and processes RETURNVALUE stream.
This stream is used to send OUTPUT parameters from RPC to client.
Stream format url: http://msdn.microsoft.com/en-us/library/dd303881.aspx
"""
r = self._reader
if IS_TDS72_PLUS(self):
ordinal = r.get_usmallint()
else:
r.get_usmallint() # ignore size
ordinal = self._out_params_indexes[self.return_value_index]
name = r.read_ucs2(r.get_byte())
r.get_byte() # 1 - OUTPUT of sp, 2 - result of udf
param = Column()
param.column_name = name
self.get_type_info(param)
param.value = param.type.read(r)
self.output_params[ordinal] = param
self.return_value_index += 1
self.param_info = None
self.has_status = False
self.ret_status = False
self.rows_affected = TDS_NO_COUNT
self.more_rows = True
self.row = [None] * num_cols
self.res_info = info = _Results()
#
# loop through the columns populating COLINFO struct from
# server response
#
#logger.debug("setting up {0} columns".format(num_cols))
header_tuple = []
for col in range(num_cols):
curcol = Column()
info.columns.append(curcol)
self.get_type_info(curcol)
#
# under 7.0 lengths are number of characters not
# number of bytes... read_ucs2 handles this
#
curcol.column_name = r.read_ucs2(r.get_byte())
precision = curcol.type.precision if hasattr(curcol.type, 'precision') else None
scale = curcol.type.scale if hasattr(curcol.type, 'scale') else None
size = curcol.type._size if hasattr(curcol.type, '_size') else None
header_tuple.append((curcol.column_name, curcol.type.get_typeid(), None, size, precision, scale, curcol.column_nullable))
info.description = tuple(header_tuple)
return info
def get_type_info(self, curcol):
""" Reads TYPE_INFO structure (http://msdn.microsoft.com/en-us/library/dd358284.aspx)
:param curcol: An instance of :class:`Column` that will receive read information
"""
r = self._reader
# User defined data type of the column
curcol.column_usertype = r.get_uint() if IS_TDS72_PLUS(self) else r.get_usmallint()
curcol.flags = r.get_usmallint() # Flags
curcol.column_nullable = curcol.flags & Column.fNullable
curcol.column_writeable = (curcol.flags & Column.fReadWrite) > 0
curcol.column_identity = (curcol.flags & Column.fIdentity) > 0
type_id = r.get_byte()
type_class = self._tds._type_map.get(type_id)
if not type_class:
raise InterfaceError('Invalid type id', type_id)
curcol.type = type_class.from_stream(r)
Function guesses type of the parameter from the type of value.
Value can also be of a special types:
- An instance of :class:`Column`, in which case it is just returned.
- An instance of :class:`output`, in which case parameter will become
an output parameter.
- A singleton :var:`default`, in which case default value will be passed
into a stored proc.
:param name: Name of the parameter, will populate column_name property of returned column.
:param value: Value of the parameter, also used to guess the type of parameter.
:return: An instance of :class:`Column`
"""
if isinstance(value, Column):
value.column_name = name
return value
column = Column()
column.column_name = name
column.flags = 0
if isinstance(value, output):
column.flags |= fByRefValue
value = value.value
if value is default:
column.flags = fDefaultValue
value = None
column.value = value
if value is None:
column.type = self.conn.NVarChar(1, collation=self.conn.collation)
elif isinstance(value, bool):
column.type = BitN()