twitter/stream.py

   1 try:
   2     import urllib.request as urllib_request
   3     import urllib.error as urllib_error
   4     import io
   5 except ImportError:
   6     import urllib2 as urllib_request
   7     import urllib2 as urllib_error
   8 import json
   9
  10 from .api import TwitterCall, wrap_response
  11
  12 class TwitterJSONIter(object):
  13
  14     def __init__(self, handle, uri, arg_data):
  15         self.decoder = json.JSONDecoder()
  16         self.handle = handle
  17         self.buf = b""
  18
  19     def __iter__(self):
  20         while True:
  21             # this is a non-blocking read (ie, it will return if any data is available)
  22             self.buf += self.handle.fp._sock.fp._sock.recv(1024)
  23             try:
  24                 utf8_buf = self.buf.decode('utf8').lstrip()
  25                 res, ptr = self.decoder.raw_decode(utf8_buf)
  26                 self.buf = utf8_buf[ptr:].encode('utf8')
  27                 yield wrap_response(res, self.handle.headers)
  28             except ValueError as e:
  29                 continue
  30             except urllib_error.HTTPError as e:
  31                 raise TwitterHTTPError(e, uri, self.format, arg_data)
  32
  33 class TwitterStreamCall(TwitterCall):
  34     def _handle_response(self, req, uri, arg_data):
  35         handle = urllib_request.urlopen(req,)
  36         return iter(TwitterJSONIter(handle, uri, arg_data))
  37
  38 class TwitterStream(TwitterStreamCall):
  39     """
  40     Interface to the Twitter Stream API (stream.twitter.com). This can
  41     be used pretty much the same as the Twitter class except the
  42     result of calling a method will be an iterator that yields objects
  43     decoded from the stream. For example::
  44
  45         twitter_stream = TwitterStream(auth=UserPassAuth('joe', 'joespassword'))
  46         iterator = twitter_stream.statuses.sample()
  47
  48         for tweet in iterator:
  49             ...do something with this tweet...
  50
  51     The iterator will yield tweets forever and ever (until the stream
  52     breaks at which point it raises a TwitterHTTPError.)
  53     """
  54     def __init__(
  55         self, domain="stream.twitter.com", secure=True, auth=None,
  56         api_version='1'):
  57         uriparts = ()
  58         uriparts += (str(api_version),)
  59
  60         TwitterStreamCall.__init__(
  61             self, auth=auth, format="json", domain=domain,
  62             callable_cls=TwitterStreamCall,
  63             secure=secure, uriparts=uriparts)