forked from python-twitter-tools/twitter
-
Notifications
You must be signed in to change notification settings - Fork 4
/
api.py
150 lines (119 loc) · 4.45 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
from base64 import b64encode
from urllib import urlencode
import httplib
from exceptions import Exception
def _py26OrGreater():
import sys
return sys.hexversion > 0x20600f0
if _py26OrGreater():
import json
else:
import simplejson as json
class TwitterError(Exception):
"""
Exception thrown by the Twitter object when there is an
error interacting with twitter.com.
"""
pass
class TwitterCall(object):
def __init__(self, username, password, format, uri=""):
self.username = username
self.password = password
self.format = format
self.uri = uri
def __getattr__(self, k):
try:
return object.__getattr__(self, k)
except AttributeError:
return TwitterCall(
self.username, self.password, self.format,
self.uri + "/" + k)
def __call__(self, **kwargs):
method = "GET"
if (self.uri.endswith('new')
or self.uri.endswith('update')
or self.uri.endswith('create')
or self.uri.endswith('destroy')):
method = "POST"
encoded_kwargs = urlencode(kwargs.items())
argStr = ""
if kwargs and (method == "GET"):
argStr = "?" + encoded_kwargs
headers = {}
if (self.username):
headers["Authorization"] = "Basic " + b64encode("%s:%s" %(
self.username, self.password))
if method == "POST":
headers["Content-type"] = "application/x-www-form-urlencoded"
headers["Content-length"] = len(encoded_kwargs)
domain = "twitter.com"
if (
self.uri.startswith('/search') or
self.uri.startswith('/trends')
):
domain = "search.twitter.com"
c = httplib.HTTPConnection(domain)
try:
c.putrequest(method, "%s.%s%s" %(
self.uri, self.format, argStr))
for item in headers.iteritems():
c.putheader(*item)
c.endheaders()
if method == "POST":
c.send(encoded_kwargs)
r = c.getresponse()
if (r.status == 304):
return []
elif (r.status != 200):
raise TwitterError("Twitter sent status %i: %s" %(
r.status, r.read()))
if "json" == self.format:
return json.loads(r.read())
else:
return r.read()
finally:
c.close()
class Twitter(TwitterCall):
"""
The minimalist yet fully featured Twitter API class.
Get RESTful data by accessing members of this class. The result
is decoded python objects (lists and dicts).
The Twitter API is documented here:
http://groups.google.com/group/twitter-development-talk/web/api-documentation
Examples::
twitter = Twitter("hello@foo.com", "password123")
# Get the public timeline
twitter.statuses.public_timeline()
# Get a particular friend's timeline
twitter.statuses.friends_timeline(id="billybob")
# Also supported (but totally weird)
twitter.statuses.friends_timeline.billybob()
# Send a direct message
twitter.direct_messages.new(
user="billybob",
text="I think yer swell!")
Using the data returned::
Twitter API calls return decoded JSON. This is converted into
a bunch of Python lists, dicts, ints, and strings. For example,
x = twitter.statuses.public_timeline()
# The first 'tweet' in the timeline
x[0]
# The screen name of the user who wrote the first 'tweet'
x[0]['user']['screen_name']
Getting raw XML data::
If you prefer to get your Twitter data in XML format, pass
format="xml" to the Twitter object when you instantiate it:
twitter = Twitter(format="xml")
The output will not be parsed in any way. It will be a raw string
of XML.
"""
def __init__(self, email=None, password=None, format="json"):
"""
Create a new twitter API connector using the specified
credentials (email and password). Format specifies the output
format ("json" (default) or "xml").
"""
if (format not in ("json", "xml")):
raise TwitterError("Unknown data format '%s'" %(format))
TwitterCall.__init__(self, email, password, format)
__all__ = ["Twitter", "TwitterError"]