4

I'm using a generator function from the requests module in a QT-Application, pretty much the same as in the requests-streaming example:

import json
import requests

def get_stream():
    r = requests.get('http://httpbin.org/stream/20', stream=True)
    for line in r.iter_lines():
        if line:
            yield json.loads(line)

def consume_stream():
    for message in get_stream():
       #do something

However, when there is no incoming response (f.e. irregularly incoming tweets from Twitters Streaming API), the generator get_stream will block the consume_stream method.

This might occur in any situation where a generator does not yield immediately, but hast to wait for incoming messages etc., and therefore blocks the consumer.

Is there any pattern in Python where you can consume a generator in a non-blocking way, i.e. if the generator yields, process it's results, otherwise do something else until the next results are occuring?

Improve this question
2
  • 2
    How exactly would you know to continue consuming the generator? Commented Nov 1, 2013 at 22:36
  • Sorry for the late reply: the consume_stream-method basically updates a GUI-Widget (TreeModel) and a "Abort"-Button is available. Right now, this buttons block when the get_stream does not yield anything (e.g. waiting for incoming messages/tweets). I can't modify the iter_lines-method, but I´d like to yield something like "Waiting for tweets.." if the are no incoming tweets, and yield the tweet as soon as there is a new one arriving. So it comes down to the question wether one can control the consumption of a generator inside the consumers for-loop or has to wait.. Commented Nov 4, 2013 at 20:06

4 Answers 4

Reset to default
2

Take a look at the producer-consumer pattern. It is commonly implemented in python using a Queue.

The producer, typically running in a thread or another process (Queue supports either), simply puts messages in the queue. The consumer, whenever it feels like it, pops messages from the queue. This operation supports a timeout argument.

Improve this answer
Sign up to request clarification or add additional context in comments.

Comments

1

As Simeon is asking in the comment, it cannot work as simple as you describe it in your example. There are quite some details you have to take care of. There are different solutions, which make more or less sense, depending on your use case. You don't give much details about what you really want to do, so I will just send you to http://twistedmatrix.com/trac/wiki/QTReactor as an example. There a different solutions/frameworks which implement async message queues. And I think, that's what you are looking for.

Improve this answer

Comments

1

If you control the generator function, one solution would be to have it throw an exception after a timeout period. Perhaps something along the lines of:

def get_stream(timeout=None):
    while message=read_message(timeout=timout):
        yield message

Then have read_message throw a TimeOutException or something if a timeout condition occurs.

Of course, you still have to deal with the logistics of when/how to retry/resume.

Improve this answer

Comments

0

You could use async generators as of python 3.6 https://www.python.org/dev/peps/pep-0525/

import json
import requests

async def get_stream():
    r = requests.get('http://httpbin.org/stream/20', stream=True)
    for line in r.iter_lines():
        if line:
            yield json.loads(line)

async def consume_stream():
   await for message in get_stream():
       #do something
Improve this answer

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.