For streams, makes sleeping interruptible.
@@ -369,6 +369,21 @@ def catch_gzip_errors(f):
return new_f
+def interruptible_sleep(t, event=None):
+ """
+ Sleeps for a specified duration, optionally stopping early for event.
+
+ Returns True if interrupted
+ """
+ logging.info("sleeping %s", t)
+ total_t = 0
+ while total_t < t and (event is None or not event.is_set()):
+ time.sleep(1)
+ total_t += 1
+
+ return True if event and event.is_set() else False
+
+
class Twarc(object):
"""
Twarc allows you retrieve data from the Twitter API. Each method
@@ -613,7 +628,7 @@ class Twarc(object):
errors = 0
for line in resp.iter_lines(chunk_size=1024):
if event and event.is_set():
- logging.info("Stopping filter")
+ logging.info("stopping filter")
# Explicitly close response
resp.close()
return
@@ -631,23 +646,23 @@ class Twarc(object):
logging.warn("too many errors")
raise e
if e.response.status_code == 420:
- t = errors * 60
- logging.info("sleeping %s", t)
- time.sleep(t)
+ if interruptible_sleep(errors * 60, event):
+ logging.info("stopping filter")
+ return
else:
- t = errors * 5
- logging.info("sleeping %s", t)
- time.sleep(t)
+ if interruptible_sleep(errors * 5, event):
+ logging.info("stopping filter")
+ return
except Exception as e:
errors += 1
logging.error("caught exception %s on %s try", e, errors)
if self.http_errors and errors == self.http_errors:
logging.warn("too many exceptions")
raise e
- t = errors * 1
logging.error(e)
- logging.info("sleeping %s", t)
- time.sleep(t)
+ if interruptible_sleep(errors, event):
+ logging.info("stopping filter")
+ return
def sample(self, event=None):
"""
@@ -669,7 +684,7 @@ class Twarc(object):
errors = 0
for line in resp.iter_lines(chunk_size=512):
if event and event.is_set():
- logging.info("Stopping sample")
+ logging.info("stopping sample")
# Explicitly close response
resp.close()
return
@@ -687,22 +702,23 @@ class Twarc(object):
logging.warn("too many errors")
raise e
if e.response.status_code == 420:
- t = errors * 60
- logging.info("sleeping %s", t)
- time.sleep(t)
+ if interruptible_sleep(errors * 60, event):
+ logging.info("stopping filter")
+ return
else:
- t = errors * 5
- logging.info("sleeping %s", t)
- time.sleep(t)
+ if interruptible_sleep(errors * 5, event):
+ logging.info("stopping filter")
+ return
+
except Exception as e:
errors += 1
logging.error("caught exception %s on %s try", e, errors)
if self.http_errors and errors == self.http_errors:
logging.warn("too many errors")
raise e
- t = errors * 1
- logging.info("sleeping %s", t)
- time.sleep(t)
+ if interruptible_sleep(errors, event):
+ logging.info("stopping filter")
+ return
def dehydrate(self, iterator):
"""
