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_rate_limit(resource, limit=100, interval=60, unit=1):
return RateLimit(conn, resource, limit=limit, interval=interval, unit=unit)
def flush(cls, crawler):
for stage in crawler.stages:
conn.delete(make_key(crawler, stage))
for run_id in cls.run_ids(crawler):
conn.delete(make_key(crawler, run_id))
conn.delete(make_key(crawler, run_id, "start"))
conn.delete(make_key(crawler, run_id, "end"))
conn.delete(make_key(crawler, run_id, "total_ops"))
conn.delete(make_key(crawler, "runs"))
conn.delete(make_key(crawler, "current_run"))
conn.delete(make_key(crawler, "total_ops"))
conn.delete(make_key(crawler, "last_run"))
conn.delete(make_key(crawler, "runs_abort"))
def get_worker():
return MemoriousWorker(conn=conn)
def event_list(cls, key, start, end):
results = []
events = conn.lrange(key, start, end)
if events is None:
return results
for event in events:
result = load_json(event)
result["timestamp"] = unpack_datetime(result['timestamp'])
results.append(result)
return results