Skip to content

Instantly share code, notes, and snippets.

@zmwangx
Last active January 23, 2021 15:28
Show Gist options
  • Save zmwangx/0dcbdbe6f67f3540dd73e777ef1b2a89 to your computer and use it in GitHub Desktop.
Save zmwangx/0dcbdbe6f67f3540dd73e777ef1b2a89 to your computer and use it in GitHub Desktop.
Python requests.iter_content with timeout
#!/usr/bin/env python3
import sys
import eventlet
eventlet.monkey_patch(socket=True)
from eventlet.timeout import Timeout
import requests
def monkeypatch_method(cls):
def decorator(func):
setattr(cls, func.__name__, func)
return func
return decorator
@monkeypatch_method(requests.Response)
def iter_content_with_timeout(self, **kwargs):
if 'timeout' in kwargs:
timeout = kwargs.pop('timeout')
else:
raise TypeError('timeout not provided')
it = self.iter_content(**kwargs)
try:
while True:
with Timeout(timeout):
yield next(it)
finally:
it.close()
def fetch():
r = requests.get('http://127.0.0.1:5000/', stream=True)
assert r.status_code == 200
try:
for chunk in r.iter_content_with_timeout(chunk_size=1, timeout=1.75):
sys.stdout.buffer.write(chunk)
sys.stdout.flush()
except Timeout:
print('no content received for 1.75 seconds')
def main():
try:
fetch()
except KeyboardInterrupt:
pass
finally:
print()
if __name__ == '__main__':
main()
#!/usr/bin/env python3
import random
import time
import flask
app = flask.Flask(__name__)
@app.route('/')
def slow_stream():
def generate():
while True:
timeout = random.randint(1, 20) / 10
if timeout != 2:
print('sleeping for %.1f seconds...' % timeout)
time.sleep(timeout)
yield '.'
else:
print('wrapping up...')
return
return flask.Response(generate(), mimetype='text/plain')
def main():
app.run(host='127.0.0.1', port=5000, debug=True, threaded=True)
if __name__ == '__main__':
main()
@zmwangx
Copy link
Author

zmwangx commented Apr 23, 2019

Unfortunately, eventlet is pretty broken on Python 3.7.x.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment