Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def pop_data(self, key):
query = self.kv().where(self.KV.key == key)
if self.database.for_update:
query = query.for_update()
with self.database.atomic():
try:
kv = query.get()
except self.KV.DoesNotExist:
return EmptyData
else:
dq = self.KV.delete().where(
(self.KV.queue == self.name) &
(self.KV.key == key))
return kv.value if dq.execute() == 1 else EmptyData
def pop_data(self, key): return EmptyData
def has_data_for_key(self, key): return False
def pop_data(self, key):
try:
kv = self.kv().where(KeyValue.key == key).get()
except KeyValue.DoesNotExist:
return EmptyData
else:
dq = KeyValue.delete().where(
(KeyValue.queue == self.name) &
(KeyValue.key == key))
return kv.value if dq.execute() == 1 else EmptyData
def _get(self, preserve=False):
task_id = self.task.task_id
if self._result is EmptyData:
res = self.huey._get_data(task_id, peek=preserve)
if res is not EmptyData:
self._result = pickle.loads(res)
return self._result
else:
return res
else:
return self._result
def pop_data(self, key):
return self._results.pop(key, EmptyData)
def peek_data(self, key):
result = self.kt.get_bytes(self.prefix_key(key), self._db)
return EmptyData if result is None else result
def pop_data(self, key):
pipe = self.conn.pipeline()
pipe.hexists(self.result_key, key)
pipe.hget(self.result_key, key)
pipe.hdel(self.result_key, key)
exists, val, n = pipe.execute()
return EmptyData if not exists else val
def pop_data(self, key):
if self.expire_time is not None:
return self.peek_data(key)
result = self.kt.seize(self.prefix_key(key), self._db)
return EmptyData if result is None else result
def peek_data(self, key):
return self._results.get(key, EmptyData)
def peek_data(self, key):
try:
kv = self.kv(self.KV.value).where(self.KV.key == key).get()
except self.KV.DoesNotExist:
return EmptyData
else:
return kv.value