Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def Reference_pickler(data):
try:
marshal_dump = marshal.dumps(long(data))
except AttributeError:
marshal_dump = "i%s" % struct.pack("
def __getattr__(self, key, default=None):
if key == "id":
return long(self)
if key in self._table:
self.__allocate()
if self._record:
# to deal with case self.update_record()
return self._record.get(key, default)
else:
return None
class Serializable(object):
def as_dict(self, flat=False, sanitize=True):
return self.__dict__
def as_xml(self, sanitize=True):
return serializers.xml(self.as_dict(flat=True, sanitize=sanitize))
def as_json(self, sanitize=True):
return serializers.json(self.as_dict(flat=True, sanitize=sanitize))
def as_yaml(self, sanitize=True):
return serializers.yaml(self.as_dict(flat=True, sanitize=sanitize))
class Reference(long):
def __allocate(self):
if not self._record:
self._record = self._table[long(self)]
if not self._record:
raise RuntimeError(
"Using a recursive select but encountered a broken "
+ "reference: %s %d" % (self._table, long(self))
)
def __getattr__(self, key, default=None):
if key == "id":
return long(self)
if key in self._table:
self.__allocate()
if self._record:
# to deal with case self.update_record()
value = long(value)
elif field.type.startswith("list:string"):
value = bar_decode_string(value)
elif field.type.startswith(list_reference_s):
ref_table = field.type[len(list_reference_s) :].strip()
if id_map is not None:
value = [
id_map[ref_table][long(v)] for v in bar_decode_string(value)
]
else:
value = [v for v in bar_decode_string(value)]
elif field.type.startswith("list:"):
value = bar_decode_integer(value)
elif id_map and field.type.startswith("reference"):
try:
value = id_map[field.type[9:].strip()][long(value)]
except KeyError:
pass
elif id_offset and field.type.startswith("reference"):
try:
value = id_offset[field.type[9:].strip()] + long(value)
except KeyError:
pass
return value
def lastrowid(self, table):
if self._last_insert:
return long(self.cursor.fetchone()[0])
sequence_name = table._sequence_name
self.execute("SELECT currval(%s);" % self.adapt(sequence_name))
return long(self.cursor.fetchone()[0])
def lastrowid(self, table):
sequence_name = table._sequence_name
self.execute("SELECT gen_id(%s, 0) FROM rdb$database" % sequence_name)
return long(self.cursor.fetchone()[0])
def __allocate(self):
if not self._record:
self._record = self._table[long(self)]
if not self._record:
raise RuntimeError(
"Using a recursive select but encountered a broken "
+ "reference: %s %d" % (self._table, long(self))
)
def none_exception(value):
"""
Returns a cleaned up value that can be used for csv export:
- unicode text is encoded as such
- None values are replaced with the given representation (default )
"""
if value is None:
return null
elif PY2 and isinstance(value, unicode):
return value.encode("utf8")
elif isinstance(value, Reference):
return long(value)
elif hasattr(value, "isoformat"):
return value.isoformat()[:19].replace("T", " ")
elif isinstance(value, (list, tuple)): # for type='list:..'
return bar_encode(value)
return value
for x in range(24)]), 0)
elif arg.isalnum():
if not arg.startswith("0x"):
arg = "0x%s" % arg
try:
arg = int(arg, 0)
except ValueError as e:
raise ValueError(
"invalid objectid argument string: %s" % e)
else:
raise ValueError("Invalid objectid argument string. " +
"Requires an integer or base 16 value")
elif isinstance(arg, self.ObjectId):
return arg
elif isinstance(arg, (Row, Reference)):
return self.object_id(long(arg['id']))
elif not isinstance(arg, (int, long)):
raise TypeError(
"object_id argument must be of type ObjectId or an objectid " +
"representable integer (type %s)" % type(arg))
hexvalue = hex(arg)[2:].rstrip('L').zfill(24)
return self.ObjectId(hexvalue)