--- a/script/lib/tweetstream/CHANGELOG Tue Jan 18 18:25:18 2011 +0100
+++ b/script/lib/tweetstream/CHANGELOG Fri Jul 01 10:15:32 2011 +0200
@@ -41,4 +41,14 @@
- Removed a spurious print statement left over from debugging
- Introduced common base class for all tweetstream exceptions
- - Make sure we raise a sensible error on 404. Include url in desc of that error
\ No newline at end of file
+ - Make sure we raise a sensible error on 404. Include url in desc of that error
+
+0.3.6
+
+ - Added LocationStream class for filtering on location bounding boxes.
+
+1.0.0
+
+ - Changed API to match latest twitter endpoints. This adds SampleStream and
+ FilterStream and deprecates TweetStream, FollowStream, LocationStream,
+ TrackStream and ReconnectingTweetStream.
--- a/script/lib/tweetstream/README Tue Jan 18 18:25:18 2011 +0100
+++ b/script/lib/tweetstream/README Fri Jul 01 10:15:32 2011 +0200
@@ -7,19 +7,19 @@
Introduction
------------
-tweetstream provides a class, TweetStream, that can be used to get
-tweets from Twitter's streaming API. An instance of the class can be used as
-an iterator. In addition to fetching tweets, the object keeps track of
-the number of tweets collected and the rate at which tweets are received.
+tweetstream provides two classes, SampleStream and FollowStream, that can be
+used to get tweets from Twitter's streaming API. An instance of one of the
+classes can be used as an iterator. In addition to fetching tweets, the
+object keeps track of the number of tweets collected and the rate at which
+tweets are received.
-Subclasses are available for accessing the "track" and "follow" streams
-as well.
-
-There's also a ReconnectingTweetStream class that handles automatic
-reconnecting.
+SampleStream delivers a sample of all tweets. FilterStream delivers
+tweets that match one or more criteria. Note that it's not possible
+to get all tweets without access to the "firehose" stream, which
+is not currently avaliable to the public.
Twitter's documentation about the streaming API can be found here:
-http://apiwiki.twitter.com/Streaming-API-Documentation .
+http://dev.twitter.com/pages/streaming_api_methods .
**Note** that the API is blocking. If for some reason data is not immediatly
available, calls will block until enough data is available to yield a tweet.
@@ -27,9 +27,9 @@
Examples
--------
-Printing all incomming tweets:
+Printing incoming tweets:
->>> stream = tweetstream.TweetStream("username", "password")
+>>> stream = tweetstream.SampleStream("username", "password")
>>> for tweet in stream:
... print tweet
@@ -37,7 +37,7 @@
The stream object can also be used as a context, as in this example that
prints the author for each tweet as well as the tweet count and rate:
->>> with tweetstream.TweetStream("username", "password") as stream
+>>> with tweetstream.SampleStream("username", "password") as stream
... for tweet in stream:
... print "Got tweet from %-16s\t( tweet %d, rate %.1f tweets/sec)" % (
... tweet["user"]["screen_name"], stream.count, stream.rate )
@@ -53,36 +53,35 @@
... except tweetstream.ConnectionError, e:
... print "Disconnected from twitter. Reason:", e.reason
-To get tweets that relate to specific terms, use the TrackStream:
+To get tweets that match specific criteria, use the FilterStream. FilterStreams
+take three keyword arguments: "locations", "follow" and "track".
+
+Locations are a list of bounding boxes in which geotagged tweets should originate.
+The argument should be an iterable of longitude/latitude pairs.
+
+Track specifies keywords to track. The argument should be an iterable of
+strings.
+
+Follow returns statuses that reference given users. Argument should be an iterable
+of twitter user IDs. The IDs are userid ints, not the screen names.
>>> words = ["opera", "firefox", "safari"]
->>> with tweetstream.TrackStream("username", "password", words) as stream
+>>> people = [123,124,125]
+>>> locations = ["-122.75,36.8", "-121.75,37.8"]
+>>> with tweetstream.FilterStream("username", "password", track=words,
+... follow=people, locations=locations) as stream
... for tweet in stream:
... print "Got interesting tweet:", tweet
-To get only tweets from a set of users, use the FollowStream. The following
-would get tweets for user 1, 42 and 8675309
->>> users = [1, 42, 8675309]
->>> with tweetstream.FollowStream("username", "password", users) as stream
-... for tweet in stream:
-... print "Got tweet from:", tweet["user"]["screen_name"]
-
-
-Simple tweet fetcher that sends tweets to an AMQP message server using carrot:
+Deprecated classes
+------------------
->>> from carrot.messaging import Publisher
->>> from carrot.connection import AMQPConnection
->>> from tweetstream import TweetStream
->>> amqpconn = AMQPConnection(hostname="localhost", port=5672,
-... userid="test", password="test",
-... vhost="test")
->>> publisher = Publisher(connection=amqpconn,
-... exchange="tweets", routing_key="stream")
->>> with TweetStream("username", "password") as stream:
-... for tweet in stream:
-... publisher.send(tweet)
->>> publisher.close()
+tweetstream used to contain the classes TweetStream, FollowStream, TrackStream
+LocationStream and ReconnectingTweetStream. These were deprecated when twitter
+changed its API end points. The same functionality is now available in
+SampleStream and FilterStream. The deprecated methods will emit a warning when
+used, but will remain functional for a while longer.
Changelog
@@ -98,6 +97,12 @@
requests, please report them in the project site issue tracker. Patches are
also very welcome.
+Contributors
+------------
+
+- Rune Halvorsen
+- Christopher Schierkolk
+
License
-------
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/script/lib/tweetstream/conftest.py Fri Jul 01 10:15:32 2011 +0200
@@ -0,0 +1,10 @@
+# content of conftest.py
+
+import pytest
+def pytest_addoption(parser):
+ parser.addoption("--runslow", action="store_true",
+ help="run slow tests")
+
+def pytest_runtest_setup(item):
+ if 'slow' in item.keywords and not item.config.getvalue("runslow"):
+ pytest.skip("need --runslow option to run")
\ No newline at end of file
--- a/script/lib/tweetstream/setup.py Tue Jan 18 18:25:18 2011 +0100
+++ b/script/lib/tweetstream/setup.py Fri Jul 01 10:15:32 2011 +0200
@@ -3,7 +3,7 @@
author = "Rune Halvorsen"
email = "runefh@gmail.com"
-version = "0.3.5"
+version = "1.0.0"
homepage = "http://bitbucket.org/runeh/tweetstream/"
setup(name='tweetstream',
--- a/script/lib/tweetstream/tests/test_tweetstream.py Tue Jan 18 18:25:18 2011 +0100
+++ b/script/lib/tweetstream/tests/test_tweetstream.py Fri Jul 01 10:15:32 2011 +0200
@@ -3,52 +3,62 @@
import time
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
-from nose.tools import assert_raises
-from tweetstream import TweetStream, FollowStream, TrackStream
-from tweetstream import ConnectionError, AuthenticationError
+from tweetstream import TweetStream, FollowStream, TrackStream, LocationStream
+from tweetstream import ConnectionError, AuthenticationError, SampleStream
+
+import pytest
+from pytest import raises
+slow = pytest.mark.slow
from servercontext import test_server
single_tweet = r"""{"in_reply_to_status_id":null,"in_reply_to_user_id":null,"favorited":false,"created_at":"Tue Jun 16 10:40:14 +0000 2009","in_reply_to_screen_name":null,"text":"record industry just keeps on amazing me: http:\/\/is.gd\/13lFo - $150k per song you've SHARED, not that somebody has actually DOWNLOADED.","user":{"notifications":null,"profile_background_tile":false,"followers_count":206,"time_zone":"Copenhagen","utc_offset":3600,"friends_count":191,"profile_background_color":"ffffff","profile_image_url":"http:\/\/s3.amazonaws.com\/twitter_production\/profile_images\/250715794\/profile_normal.png","description":"Digital product developer, currently at Opera Software. My tweets are my opinions, not those of my employer.","verified_profile":false,"protected":false,"favourites_count":0,"profile_text_color":"3C3940","screen_name":"eiriksnilsen","name":"Eirik Stridsklev N.","following":null,"created_at":"Tue May 06 12:24:12 +0000 2008","profile_background_image_url":"http:\/\/s3.amazonaws.com\/twitter_production\/profile_background_images\/10531192\/160x600opera15.gif","profile_link_color":"0099B9","profile_sidebar_fill_color":"95E8EC","url":"http:\/\/www.stridsklev-nilsen.no\/eirik","id":14672543,"statuses_count":506,"profile_sidebar_border_color":"5ED4DC","location":"Oslo, Norway"},"id":2190767504,"truncated":false,"source":"<a href=\"http:\/\/widgets.opera.com\/widget\/7206\">Twitter Opera widget<\/a>"}"""
-def test_bad_auth():
+def parameterized(funcarglist):
+ def wrapper(function):
+ function.funcarglist = funcarglist
+ return function
+ return wrapper
+
+def pytest_generate_tests(metafunc):
+ for funcargs in getattr(metafunc.function, 'funcarglist', ()):
+ metafunc.addcall(funcargs=funcargs)
+
+
+streamtypes = [
+ dict(cls=TweetStream, args=[], kwargs=dict()),
+ dict(cls=SampleStream, args=[], kwargs=dict()),
+ dict(cls=FollowStream, args=[[1, 2, 3]], kwargs=dict()),
+ dict(cls=TrackStream, args=["opera"], kwargs=dict()),
+ dict(cls=LocationStream, args=["123,4321"], kwargs=dict())
+]
+
+
+@parameterized(streamtypes)
+def test_bad_auth(cls, args, kwargs):
"""Test that the proper exception is raised when the user could not be
authenticated"""
def auth_denied(request):
request.send_error(401)
- with test_server(handler=auth_denied, methods=("post", "get"),
- port="random") as server:
- stream = TweetStream("foo", "bar", url=server.baseurl)
- assert_raises(AuthenticationError, stream.next)
-
- stream = FollowStream("foo", "bar", [1, 2, 3], url=server.baseurl)
- assert_raises(AuthenticationError, stream.next)
-
- stream = TrackStream("foo", "bar", ["opera"], url=server.baseurl)
- assert_raises(AuthenticationError, stream.next)
+ with test_server(handler=auth_denied, methods=("post", "get"), port="random") as server:
+ stream = cls("user", "passwd", *args, url=server.baseurl)
-def test_404_url():
+@parameterized(streamtypes)
+def test_404_url(cls, args, kwargs):
"""Test that the proper exception is raised when the stream URL can't be
found"""
def not_found(request):
request.send_error(404)
- with test_server(handler=not_found, methods=("post", "get"),
- port="random") as server:
- stream = TweetStream("foo", "bar", url=server.baseurl)
- assert_raises(ConnectionError, stream.next)
-
- stream = FollowStream("foo", "bar", [1, 2, 3], url=server.baseurl)
- assert_raises(ConnectionError, stream.next)
-
- stream = TrackStream("foo", "bar", ["opera"], url=server.baseurl)
- assert_raises(ConnectionError, stream.next)
+ with test_server(handler=not_found, methods=("post", "get"), port="random") as server:
+ stream = cls("user", "passwd", *args, url=server.baseurl)
-def test_bad_content():
+@parameterized(streamtypes)
+def test_bad_content(cls, args, kwargs):
"""Test error handling if we are given invalid data"""
def bad_content(request):
for n in xrange(10):
@@ -58,19 +68,16 @@
yield "[1,2, I need no stinking close brace"
yield "[1,2,3]"
- def do_test(klass, *args):
- with test_server(handler=bad_content, methods=("post", "get"),
- port="random") as server:
- stream = klass("foo", "bar", *args, url=server.baseurl)
+
+ with raises(ConnectionError):
+ with test_server(handler=bad_content, methods=("post", "get"), port="random") as server:
+ stream = cls("user", "passwd", *args, url=server.baseurl)
for tweet in stream:
pass
- assert_raises(ConnectionError, do_test, TweetStream)
- assert_raises(ConnectionError, do_test, FollowStream, [1, 2, 3])
- assert_raises(ConnectionError, do_test, TrackStream, ["opera"])
-
-def test_closed_connection():
+@parameterized(streamtypes)
+def test_closed_connection(cls, args, kwargs):
"""Test error handling if server unexpectedly closes connection"""
cnt = 1000
def bad_content(request):
@@ -79,31 +86,23 @@
# strcuture, only checking that it's parsable
yield "[1,2,3]"
- def do_test(klass, *args):
- with test_server(handler=bad_content, methods=("post", "get"),
- port="random") as server:
- stream = klass("foo", "bar", *args, url=server.baseurl)
+ with raises(ConnectionError):
+ with test_server(handler=bad_content, methods=("post", "get"), port="random") as server:
+ stream = cls("foo", "bar", *args, url=server.baseurl)
for tweet in stream:
pass
- assert_raises(ConnectionError, do_test, TweetStream)
- assert_raises(ConnectionError, do_test, FollowStream, [1, 2, 3])
- assert_raises(ConnectionError, do_test, TrackStream, ["opera"])
+
+@parameterized(streamtypes)
+def test_bad_host(cls, args, kwargs):
+ """Test behaviour if we can't connect to the host"""
+ with raises(ConnectionError):
+ stream = cls("username", "passwd", *args, url="http://wedfwecfghhreewerewads.foo")
+ stream.next()
-def test_bad_host():
- """Test behaviour if we can't connect to the host"""
- stream = TweetStream("foo", "bar", url="http://bad.egewdvsdswefdsf.com/")
- assert_raises(ConnectionError, stream.next)
-
- stream = FollowStream("foo", "bar", [1, 2, 3], url="http://zegwefdsf.com/")
- assert_raises(ConnectionError, stream.next)
-
- stream = TrackStream("foo", "bar", ["foo"], url="http://aswefdsews.com/")
- assert_raises(ConnectionError, stream.next)
-
-
-def smoke_test_receive_tweets():
+@parameterized(streamtypes)
+def smoke_test_receive_tweets(cls, args, kwargs):
"""Receive 100k tweets and disconnect (slow)"""
total = 100000
@@ -111,20 +110,15 @@
while True:
yield single_tweet + "\n"
- def do_test(klass, *args):
- with test_server(handler=tweetsource,
- methods=("post", "get"), port="random") as server:
- stream = klass("foo", "bar", *args, url=server.baseurl)
- for tweet in stream:
- if stream.count == total:
- break
-
- do_test(TweetStream)
- do_test(FollowStream, [1, 2, 3])
- do_test(TrackStream, ["foo", "bar"])
+ with test_server(handler=tweetsource, methods=("post", "get"), port="random") as server:
+ stream = cls("foo", "bar", *args, url=server.baseurl)
+ for tweet in stream:
+ if stream.count == total:
+ break
-def test_keepalive():
+@parameterized(streamtypes)
+def test_keepalive(cls, args, kwargs):
"""Make sure we behave sanely when there are keepalive newlines in the
data recevived from twitter"""
def tweetsource(request):
@@ -142,25 +136,21 @@
yield single_tweet+"\n"
yield "\n"
- def do_test(klass, *args):
- with test_server(handler=tweetsource, methods=("post", "get"),
- port="random") as server:
- stream = klass("foo", "bar", *args, url=server.baseurl)
- try:
- for tweet in stream:
- pass
- except ConnectionError:
- assert stream.count == 3, "Got %s, wanted 3" % stream.count
- else:
- assert False, "Didn't handle keepalive"
+
+ with test_server(handler=tweetsource, methods=("post", "get"), port="random") as server:
+ stream = cls("foo", "bar", *args, url=server.baseurl)
+ try:
+ for tweet in stream:
+ pass
+ except ConnectionError:
+ assert stream.count == 3, "Got %s, wanted 3" % stream.count
+ else:
+ assert False, "Didn't handle keepalive"
- do_test(TweetStream)
- do_test(FollowStream, [1, 2, 3])
- do_test(TrackStream, ["foo", "bar"])
-
-
-def test_buffering():
+@slow
+@parameterized(streamtypes)
+def test_buffering(cls, args, kwargs):
"""Test if buffering stops data from being returned immediately.
If there is some buffering in play that might mean data is only returned
from the generator when the buffer is full. If buffer is bigger than a
@@ -175,19 +165,12 @@
for n in xrange(100):
yield single_tweet+"\n"
- def do_test(klass, *args):
- with test_server(handler=tweetsource, methods=("post", "get"),
- port="random") as server:
- stream = klass("foo", "bar", *args, url=server.baseurl)
- start = time.time()
- stream.next()
- first = time.time()
- diff = first - start
- assert diff < 1, "Getting first tweet took more than a second!"
+ with test_server(handler=tweetsource, methods=("post", "get"), port="random") as server:
+ stream = cls("foo", "bar", *args, url=server.baseurl)
+ start = time.time()
+ stream.next()
+ first = time.time()
+ diff = first - start
+ assert diff < 1, "Getting first tweet took more than a second!"
- do_test(TweetStream)
- do_test(FollowStream, [1, 2, 3])
- do_test(TrackStream, ["foo", "bar"])
-
-
--- a/script/lib/tweetstream/tweetstream/__init__.py Tue Jan 18 18:25:18 2011 +0100
+++ b/script/lib/tweetstream/tweetstream/__init__.py Fri Jul 01 10:15:32 2011 +0200
@@ -1,34 +1,18 @@
"""
Simple Twitter streaming API access
"""
-__version__ = "0.3.5"
+__version__ = "1.0.0"
__author__ = "Rune Halvorsen <runefh@gmail.com>"
__homepage__ = "http://bitbucket.org/runeh/tweetstream/"
__docformat__ = "restructuredtext"
-import urllib
-import urllib2
-import socket
-import time
-import anyjson
-
"""
- .. data:: URLS
-
- Mapping between twitter endpoint names and URLs.
-
.. data:: USER_AGENT
The default user agent string for stream objects
-
"""
-URLS = {"firehose": "http://stream.twitter.com/1/statuses/firehose.json",
- "sample": "http://stream.twitter.com/1/statuses/sample.json",
- "follow": "http://stream.twitter.com/1/statuses/filter.json",
- "track": "http://stream.twitter.com/1/statuses/filter.json"}
-
USER_AGENT = "TweetStream %s" % __version__
@@ -36,9 +20,9 @@
"""Base class for all tweetstream errors"""
pass
+
class AuthenticationError(TweetStreamError):
- """Exception raised if the username/password is not accepted
- """
+ """Exception raised if the username/password is not accepted"""
pass
@@ -54,244 +38,5 @@
return '<ConnectionError %s>' % self.reason
-class TweetStream(object):
- """A network connection to Twitters streaming API
-
- :param username: Twitter username for the account accessing the API.
- :param password: Twitter password for the account accessing the API.
-
- :keyword url: URL to connect to. This can be either an endopoint name,
- such as "sample", or a full URL. By default, the public "sample" url
- is used. All known endpoints are defined in the :URLS: attribute
-
- .. attribute:: connected
-
- True if the object is currently connected to the stream.
-
- .. attribute:: url
-
- The URL to which the object is connected
-
- .. attribute:: starttime
-
- The timestamp, in seconds since the epoch, the object connected to the
- streaming api.
-
- .. attribute:: count
-
- The number of tweets that have been returned by the object.
-
- .. attribute:: rate
-
- The rate at which tweets have been returned from the object as a
- float. see also :attr: `rate_period`.
-
- .. attribute:: rate_period
-
- The ammount of time to sample tweets to calculate tweet rate. By
- default 10 seconds. Changes to this attribute will not be reflected
- until the next time the rate is calculated. The rate of tweets vary
- with time of day etc. so it's usefull to set this to something
- sensible.
-
- .. attribute:: user_agent
-
- User agent string that will be included in the request. NOTE: This can
- not be changed after the connection has been made. This property must
- thus be set before accessing the iterator. The default is set in
- :attr: `USER_AGENT`.
-"""
-
- def __init__(self, username, password, url="sample"):
- self._conn = None
- self._rate_ts = None
- self._rate_cnt = 0
- self._username = username
- self._password = password
-
- self.rate_period = 10 # in seconds
- self.connected = False
- self.starttime = None
- self.count = 0
- self.rate = 0
- self.user_agent = USER_AGENT
- self.url = URLS.get(url, url)
-
- def __iter__(self):
- return self
-
- def __enter__(self):
- return self
-
- def __exit__(self, *params):
- self.close()
- return False
-
- def _init_conn(self):
- """Open the connection to the twitter server"""
- headers = {'User-Agent': self.user_agent}
- req = urllib2.Request(self.url, self._get_post_data(), headers)
-
- password_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
- password_mgr.add_password(None, self.url, self._username,
- self._password)
- handler = urllib2.HTTPBasicAuthHandler(password_mgr)
- opener = urllib2.build_opener(handler)
-
- try:
- self._conn = opener.open(req)
- except urllib2.HTTPError, exception:
- if exception.code == 401:
- raise AuthenticationError("Access denied")
- elif exception.code == 404:
- raise ConnectionError("URL not found: %s" % self.url)
- else: # re raise. No idea what would cause this, so want to know
- raise
- except urllib2.URLError, exception:
- raise ConnectionError(exception.reason)
-
- self.connected = True
- if not self.starttime:
- self.starttime = time.time()
- if not self._rate_ts:
- self._rate_ts = time.time()
-
- def _get_post_data(self):
- """Subclasses that need to add post data to the request can override
- this method and return post data. The data should be in the format
- returned by urllib.urlencode."""
- return None
-
- def next(self):
- """Return the next available tweet. This call is blocking!"""
- while True:
- try:
- if not self.connected:
- self._init_conn()
-
- rate_time = time.time() - self._rate_ts
- if not self._rate_ts or rate_time > self.rate_period:
- self.rate = self._rate_cnt / rate_time
- self._rate_cnt = 0
- self._rate_ts = time.time()
-
- data = self._conn.readline()
- if data == "": # something is wrong
- self.close()
- raise ConnectionError("Got entry of length 0. Disconnected")
- elif data.isspace():
- continue
-
- data = anyjson.deserialize(data)
- self.count += 1
- self._rate_cnt += 1
- return data
-
- except ValueError, e:
- self.close()
- raise ConnectionError("Got invalid data from twitter", details=data)
-
- except socket.error, e:
- self.close()
- raise ConnectionError("Server disconnected")
-
-
- def close(self):
- """
- Close the connection to the streaming server.
- """
- self.connected = False
- if self._conn:
- self._conn.close()
-
-
-class ReconnectingTweetStream(TweetStream):
- """TweetStream class that automatically tries to reconnect if the
- connecting goes down. Reconnecting, and waiting for reconnecting, is
- blocking.
-
- :param username: See :TweetStream:
-
- :param password: See :TweetStream:
-
- :keyword url: See :TweetStream:
-
- :keyword reconnects: Number of reconnects before a ConnectionError is
- raised. Default is 3
-
- :error_cb: Optional callable that will be called just before trying to
- reconnect. The callback will be called with a single argument, the
- exception that caused the reconnect attempt. Default is None
-
- :retry_wait: Time to wait before reconnecting in seconds. Default is 5
-
- """
-
- def __init__(self, username, password, url="sample",
- reconnects=3, error_cb=None, retry_wait=5):
- self.max_reconnects = reconnects
- self.retry_wait = retry_wait
- self._reconnects = 0
- self._error_cb = error_cb
- TweetStream.__init__(self, username, password, url=url)
-
- def next(self):
- while True:
- try:
- return TweetStream.next(self)
- except ConnectionError, e:
- self._reconnects += 1
- if self._reconnects > self.max_reconnects:
- raise ConnectionError("Too many retries")
-
- # Note: error_cb is not called on the last error since we
- # raise a ConnectionError instead
- if callable(self._error_cb):
- self._error_cb(e)
-
- time.sleep(self.retry_wait)
- # Don't listen to auth error, since we can't reasonably reconnect
- # when we get one.
-
-class FollowStream(TweetStream):
- """Stream class for getting tweets from followers.
-
- :param user: See TweetStream
-
- :param password: See TweetStream
-
- :param followees: Iterable containing user IDs to follow.
- ***Note:*** the user id in question is the numeric ID twitter uses,
- not the normal username.
-
- :keyword url: Like the url argument to TweetStream, except default
- value is the "follow" endpoint.
- """
-
- def __init__(self, user, password, followees, url="follow", **kwargs):
- self.followees = followees
- TweetStream.__init__(self, user, password, url=url, **kwargs)
-
- def _get_post_data(self):
- return urllib.urlencode({"follow": ",".join(map(str, self.followees))})
-
-
-class TrackStream(TweetStream):
- """Stream class for getting tweets relevant to keywords.
-
- :param user: See TweetStream
-
- :param password: See TweetStream
-
- :param keywords: Iterable containing keywords to look for
-
- :keyword url: Like the url argument to TweetStream, except default
- value is the "track" endpoint.
- """
-
- def __init__(self, user, password, keywords, url="track", **kwargs):
- self.keywords = keywords
- TweetStream.__init__(self, user, password, url=url, **kwargs)
-
- def _get_post_data(self):
- return urllib.urlencode({"track": ",".join(self.keywords)})
+from streamclasses import SampleStream, FilterStream
+from deprecated import FollowStream, TrackStream, LocationStream, TweetStream, ReconnectingTweetStream
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/script/lib/tweetstream/tweetstream/deprecated.py Fri Jul 01 10:15:32 2011 +0200
@@ -0,0 +1,81 @@
+from .streamclasses import FilterStream, SampleStream
+
+class DeprecatedStream(FilterStream):
+ def __init__(self, *args, **kwargs):
+ import warnings
+ warnings.warn("%s is deprecated. Use FilterStream instead" % self.__class__.__name__, DeprecationWarning)
+ super(DeprecatedStream, self).__init__(*args, **kwargs)
+
+
+class FollowStream(DeprecatedStream):
+ def __init__(self, username, password, follow, catchup=None, url=None):
+ super(FollowStream, self).__init__(username, password, follow=follow, catchup=catchup, url=url)
+
+
+class TrackStream(DeprecatedStream):
+ def __init__(self, username, password, track, catchup=None, url=None):
+ super(TrackStream, self).__init__(username, password, track=track, catchup=catchup, url=url)
+
+
+class LocationStream(DeprecatedStream):
+ def __init__(self, username, password, locations, catchup=None, url=None):
+ super(LocationStream, self).__init__(username, password, locations=locations, catchup=catchup, url=url)
+
+
+class TweetStream(SampleStream):
+ def __init__(self, *args, **kwargs):
+ import warnings
+ warnings.warn("%s is deprecated. Use SampleStream instead" % self.__class__.__name__, DeprecationWarning)
+ SampleStream.__init__(self, *args, **kwargs)
+
+
+class ReconnectingTweetStream(TweetStream):
+ """TweetStream class that automatically tries to reconnect if the
+ connecting goes down. Reconnecting, and waiting for reconnecting, is
+ blocking.
+
+ :param username: See :TweetStream:
+
+ :param password: See :TweetStream:
+
+ :keyword url: See :TweetStream:
+
+ :keyword reconnects: Number of reconnects before a ConnectionError is
+ raised. Default is 3
+
+ :error_cb: Optional callable that will be called just before trying to
+ reconnect. The callback will be called with a single argument, the
+ exception that caused the reconnect attempt. Default is None
+
+ :retry_wait: Time to wait before reconnecting in seconds. Default is 5
+
+ """
+
+ def __init__(self, username, password, url="sample",
+ reconnects=3, error_cb=None, retry_wait=5):
+ self.max_reconnects = reconnects
+ self.retry_wait = retry_wait
+ self._reconnects = 0
+ self._error_cb = error_cb
+ TweetStream.__init__(self, username, password, url=url)
+
+ def next(self):
+ while True:
+ try:
+ return TweetStream.next(self)
+ except ConnectionError, e:
+ self._reconnects += 1
+ if self._reconnects > self.max_reconnects:
+ raise ConnectionError("Too many retries")
+
+ # Note: error_cb is not called on the last error since we
+ # raise a ConnectionError instead
+ if callable(self._error_cb):
+ self._error_cb(e)
+
+ time.sleep(self.retry_wait)
+ # Don't listen to auth error, since we can't reasonably reconnect
+ # when we get one.
+
+
+
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/script/lib/tweetstream/tweetstream/streamclasses.py Fri Jul 01 10:15:32 2011 +0200
@@ -0,0 +1,187 @@
+import time
+import urllib
+import urllib2
+import socket
+import anyjson
+
+from . import AuthenticationError, ConnectionError, USER_AGENT
+
+class BaseStream(object):
+ """A network connection to Twitters streaming API
+
+ :param username: Twitter username for the account accessing the API.
+ :param password: Twitter password for the account accessing the API.
+ :keyword count: Number of tweets from the past to get before switching to
+ live stream.
+ :keyword url: Endpoint URL for the object. Note: you should not
+ need to edit this. It's present to make testing easier.
+
+ .. attribute:: connected
+
+ True if the object is currently connected to the stream.
+
+ .. attribute:: url
+
+ The URL to which the object is connected
+
+ .. attribute:: starttime
+
+ The timestamp, in seconds since the epoch, the object connected to the
+ streaming api.
+
+ .. attribute:: count
+
+ The number of tweets that have been returned by the object.
+
+ .. attribute:: rate
+
+ The rate at which tweets have been returned from the object as a
+ float. see also :attr: `rate_period`.
+
+ .. attribute:: rate_period
+
+ The ammount of time to sample tweets to calculate tweet rate. By
+ default 10 seconds. Changes to this attribute will not be reflected
+ until the next time the rate is calculated. The rate of tweets vary
+ with time of day etc. so it's usefull to set this to something
+ sensible.
+
+ .. attribute:: user_agent
+
+ User agent string that will be included in the request. NOTE: This can
+ not be changed after the connection has been made. This property must
+ thus be set before accessing the iterator. The default is set in
+ :attr: `USER_AGENT`.
+ """
+
+ def __init__(self, username, password, catchup=None, url=None):
+ self._conn = None
+ self._rate_ts = None
+ self._rate_cnt = 0
+ self._username = username
+ self._password = password
+ self._catchup_count = catchup
+
+ self.rate_period = 10 # in seconds
+ self.connected = False
+ self.starttime = None
+ self.count = 0
+ self.rate = 0
+ self.user_agent = USER_AGENT
+ if url: self.url = url
+
+ def __iter__(self):
+ return self
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *params):
+ self.close()
+ return False
+
+ def _init_conn(self):
+ """Open the connection to the twitter server"""
+ headers = {'User-Agent': self.user_agent}
+
+ postdata = self._get_post_data() or {}
+ if self._catchup_count:
+ postdata["count"] = self._catchup_count
+
+ poststring = urllib.urlencode(postdata) if postdata else None
+ req = urllib2.Request(self.url, poststring, headers)
+
+ password_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
+ password_mgr.add_password(None, self.url, self._username, self._password)
+ handler = urllib2.HTTPBasicAuthHandler(password_mgr)
+ opener = urllib2.build_opener(handler)
+
+ try:
+ self._conn = opener.open(req)
+ except urllib2.HTTPError, exception:
+ if exception.code == 401:
+ raise AuthenticationError("Access denied")
+ elif exception.code == 404:
+ raise ConnectionError("URL not found: %s" % self.url)
+ else: # re raise. No idea what would cause this, so want to know
+ raise
+ except urllib2.URLError, exception:
+ raise ConnectionError(exception.reason)
+
+ self.connected = True
+ if not self.starttime:
+ self.starttime = time.time()
+ if not self._rate_ts:
+ self._rate_ts = time.time()
+
+ def _get_post_data(self):
+ """Subclasses that need to add post data to the request can override
+ this method and return post data. The data should be in the format
+ returned by urllib.urlencode."""
+ return None
+
+ def next(self):
+ """Return the next available tweet. This call is blocking!"""
+ while True:
+ try:
+ if not self.connected:
+ self._init_conn()
+
+ rate_time = time.time() - self._rate_ts
+ if not self._rate_ts or rate_time > self.rate_period:
+ self.rate = self._rate_cnt / rate_time
+ self._rate_cnt = 0
+ self._rate_ts = time.time()
+
+ data = self._conn.readline()
+ if data == "": # something is wrong
+ self.close()
+ raise ConnectionError("Got entry of length 0. Disconnected")
+ elif data.isspace():
+ continue
+
+ data = anyjson.deserialize(data)
+ if 'text' in data:
+ self.count += 1
+ self._rate_cnt += 1
+ return data
+
+ except ValueError, e:
+ self.close()
+ raise ConnectionError("Got invalid data from twitter",
+ details=data)
+
+ except socket.error, e:
+ self.close()
+ raise ConnectionError("Server disconnected")
+
+ def close(self):
+ """
+ Close the connection to the streaming server.
+ """
+ self.connected = False
+ if self._conn:
+ self._conn.close()
+
+
+class SampleStream(BaseStream):
+ url = "http://stream.twitter.com/1/statuses/sample.json"
+
+
+class FilterStream(BaseStream):
+ url = "http://stream.twitter.com/1/statuses/filter.json"
+
+ def __init__(self, username, password, follow=None, locations=None,
+ track=None, catchup=None, url=None):
+ self._follow = follow
+ self._locations = locations
+ self._track = track
+ # remove follow, locations, track
+ BaseStream.__init__(self, username, password, url=url)
+
+ def _get_post_data(self):
+ postdata = {}
+ if self._follow: postdata["follow"] = ",".join([str(e) for e in self._follow])
+ if self._locations: postdata["locations"] = ",".join(self._locations)
+ if self._track: postdata["track"] = ",".join(self._track)
+ return postdata