Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if type(value) == date:
value = datetime.combine(value, time(0, 0, 0))
days = (value - cls._base_date).days
ms = value.microsecond // 1000
tm = (value.hour * 60 * 60 + value.minute * 60 + value.second) * 300 + int(round(ms * 3 / 10.0))
return cls._struct.pack(days, tm)
@classmethod
def decode(cls, days, time):
ms = int(round(time % 300 * 10 / 3.0))
secs = time // 300
return cls._base_date + timedelta(days=days, seconds=secs, milliseconds=ms)
DateTime.instance = DateTime()
class DateTimeN(BaseType):
type = SYBDATETIMN
_base_date = datetime(1900, 1, 1)
_min_date = datetime(1753, 1, 1, 0, 0, 0)
_max_date = datetime(9999, 12, 31, 23, 59, 59, 997000)
def __init__(self, size):
assert size in (4, 8)
self._size = size
self._subtype = {4: SmallDateTime.instance, 8: DateTime.instance}[size]
@classmethod
def from_stream(self, r):
size = r.get_byte()
if size not in (4, 8):
raise InterfaceError('Invalid SYBDATETIMN size', size)
def from_stream(cls, r):
return cls.instance
def write_info(self, w):
pass
def write(self, w, value):
w.put_byte(1 if value else 0)
def read(self, r):
return bool(r.get_byte())
Bit.instance = Bit()
class BitN(BaseType):
type = SYBBITN
def get_declaration(self):
return 'BIT'
@classmethod
def from_stream(cls, r):
size = r.get_byte()
if size != 1:
raise InterfaceError('Invalid BIT field size', size)
return cls()
def write_info(self, w):
w.put_byte(1)
def write(self, w, value):
pass
def get_declaration(self):
return 'SMALLMONEY'
def read(self, r):
return Decimal(r.get_int()) / 10000
def write(self, w, val):
val = int(val * 10000)
w.put_int(val)
Money4.instance = Money4()
class Money8(BaseType):
type = SYBMONEY
_struct = struct.Struct('
def get_declaration(self):
return 'TINYINT'
def write_info(self, w):
pass
def write(self, w, val):
w.put_byte(val)
def read(self, r):
return r.get_byte()
TinyInt.instance = TinyInt()
class SmallInt(BaseType):
type = SYBINT2
@classmethod
def from_stream(cls, r):
return cls()
def get_declaration(self):
return 'SMALLINT'
def write_info(self, w):
pass
def write(self, w, val):
w.put_smallint(val)
def read(self, r):
offset = r.get_smallint()
tzinfo_factory = r._session.tzinfo_factory
if tzinfo_factory is None:
from .tz import FixedOffsetTimezone
tzinfo_factory = FixedOffsetTimezone
tz = tzinfo_factory(offset)
return datetime.combine(date, time).astimezone(tz)
def read(self, r):
size = r.get_byte()
if size == 0:
return None
return self.read_fixed(r, size)
class MsDecimal(BaseType):
type = SYBDECIMAL
_max_size = 17
_bytes_per_prec = [
#
# precision can't be 0 but using a value > 0 assure no
# core if for some bug it's 0...
#
1,
5, 5, 5, 5, 5, 5, 5, 5, 5,
9, 9, 9, 9, 9, 9, 9, 9, 9, 9,
13, 13, 13, 13, 13, 13, 13, 13, 13,
17, 17, 17, 17, 17, 17, 17, 17, 17, 17,
]
if val is None:
w.put_byte(0)
else:
w.put_byte(self._size)
w.pack(self._current_struct, val)
def read(self, r):
size = r.get_byte()
if size == 0:
return None
if size not in self._valid_sizes:
raise InterfaceError('Invalid size of INTN field', size)
return r.unpack(self._struct[size])[0]
class Real(BaseType):
type = SYBREAL
@classmethod
def from_stream(cls, r):
return cls()
def get_declaration(self):
return 'REAL'
def write_info(self, w):
pass
def write(self, w, val):
w.pack(_flt4_struct, val)
def read(self, r):
size = r.get_byte()
if size == 0:
return None
if size != self._size:
raise r.session.bad_stream('Invalid SYBMONEYN size', size)
return self._subtype.read(r)
def write(self, w, val):
if val is None:
w.put_byte(0)
return
w.put_byte(self._size)
self._subtype.write(w, val)
class MsUnique(BaseType):
type = SYBUNIQUE
@classmethod
def from_stream(cls, r):
size = r.get_byte()
if size != 16:
raise InterfaceError('Invalid size of UNIQUEIDENTIFIER field')
return cls.instance
def get_declaration(self):
return 'UNIQUEIDENTIFIER'
def write_info(self, w):
w.put_byte(16)
def write(self, w, value):
val = force_unicode(val)
val, _ = self._codec.encode(val)
w.put_int8(len(val))
if len(val) > 0:
w.put_int(len(val))
w.write(val)
w.put_int(0)
def read(self, r):
r = PlpReader(r)
if r.is_null():
return None
return ''.join(iterdecode(r.chunks(), self._codec))
class NVarChar70(BaseType):
type = XSYBNVARCHAR
def __init__(self, size):
#if size <= 0 or size > 4000:
# raise DataError('Invalid size for NVARCHAR field')
self._size = size
@classmethod
def from_stream(cls, r):
size = r.get_usmallint()
return cls(size / 2)
def get_declaration(self):
return 'NVARCHAR({0})'.format(self._size)
def write_info(self, w):
def get_declaration(self):
return 'SMALLINT'
def write_info(self, w):
pass
def write(self, w, val):
w.put_smallint(val)
def read(self, r):
return r.get_smallint()
SmallInt.instance = SmallInt()
class Int(BaseType):
type = SYBINT4
@classmethod
def from_stream(cls, r):
return cls()
def get_declaration(self):
return 'INT'
def write_info(self, w):
pass
def write(self, w, val):
w.put_int(val)
def read(self, r):
r.get_collation()
r.get_usmallint()
return r.read_str(size, ucs2_codec)
def _variant_read_decimal(r, size):
prec, scale = r.unpack(Variant._decimal_info_struct)
return MsDecimal(prec=prec, scale=scale).read_fixed(r, size)
def _variant_read_binary(r, size):
r.get_usmallint()
return readall(r, size)
class Variant(BaseType):
type = SYBVARIANT
_decimal_info_struct = struct.Struct('BB')
_type_map = {
GUIDTYPE: lambda r, size: MsUnique.instance.read_fixed(r, size),
BITTYPE: lambda r, size: Bit.instance.read(r),
INT1TYPE: lambda r, size: TinyInt.instance.read(r),
INT2TYPE: lambda r, size: SmallInt.instance.read(r),
INT4TYPE: lambda r, size: Int.instance.read(r),
INT8TYPE: lambda r, size: BigInt.instance.read(r),
DATETIMETYPE: lambda r, size: DateTime.instance.read(r),
DATETIM4TYPE: lambda r, size: SmallDateTime.instance.read(r),
FLT4TYPE: lambda r, size: Real.instance.read(r),
FLT8TYPE: lambda r, size: Float.instance.read(r),
MONEYTYPE: lambda r, size: Money8.instance.read(r),