Reuse OAuth1 authorization tokens with rauth

2024/4/15 1:40:11

I have the following implementation of a twitter client using rauth (OAuth1), based on the twitter-timeline-cli.py script in the rauth examples:

from rauth.service import OAuth1Serviceclass TwitterClient:KNOWN_USERS = {'user1' : ("xxx", "yyy", "2342354"), # These should be real tokens}def __init__(self):# Get a real consumer key & secret from https://dev.twitter.com/apps/newself.twitter = OAuth1Service(name='twitter',consumer_key=TWITTER_CONSUMER_KEY,consumer_secret=TWITTER_CONSUMER_SECRET,request_token_url='https://api.twitter.com/oauth/request_token',access_token_url='https://api.twitter.com/oauth/access_token',authorize_url='https://api.twitter.com/oauth/authorize',base_url='https://api.twitter.com/1/')def authorize(self):request_token, request_token_secret = self.twitter.get_request_token()authorize_url = self.twitter.get_authorize_url(request_token)print 'Visit this URL in your browser: ' + authorize_urlpin = raw_input('Enter PIN from browser: ')return request_token, request_token_secret, pindef init_session(self, user):if user in self.KNOWN_USERS :request_token, request_token_secret, pin = self.KNOWN_USERS[user]else:request_token, request_token_secret, pin = self.authorize()session = self.twitter.get_auth_session(request_token,request_token_secret,method='POST',data={'oauth_verifier': pin})return sessiondef list_tweets(self, user):session = self.init_session(user)params = {'include_rts': 1,  # Include retweets'count': 10}       # 10 tweetsr = session.get('statuses/home_timeline.json', params=params)for i, tweet in enumerate(r.json(), 1):handle = tweet['user']['screen_name'].encode('utf-8')text = tweet['text'].encode('utf-8')print '{0}. @{1} - {2}'.format(i, handle, text)tc = TwitterClient()tc.list_tweets('user1')

The idea is that, if the user is not known, he is requested to authorize the application. If, on the other hand, the user has already authorized this application, the authorization tokens (request_token, request_token_secret, pin) should be reused (normally the tokens would be in a database; for the time being, they are hard-coded in the script)

But this is not working:

Traceback (most recent call last):File "my-twitter-timeline-cli.py", line 56, in <module>tc.list_tweets('user1')File "my-twitter-timeline-cli.py", line 43, in list_tweetssession = self.init_session(user)File "my-twitter-timeline-cli.py", line 39, in init_sessiondata={'oauth_verifier': pin})File ".../lib/python2.7/site-packages/rauth/service.py", line 326, in get_auth_session**kwargs)File ".../lib/python2.7/site-packages/rauth/service.py", line 299, in get_access_tokenprocess_token_request(r, decoder, key_token, key_token_secret)File ".../lib/python2.7/site-packages/rauth/service.py", line 25, in process_token_requestraise KeyError(PROCESS_TOKEN_ERROR.format(key=bad_key, raw=r.content))
KeyError: 'Decoder failed to handle oauth_token with data as returned by provider. A different decoder may be needed. Provider returned: <?xml version="1.0" encoding="UTF-8"?>\n<hash>\n  <error>Invalid / expired Token</error>\n  <request>/oauth/access_token</request>\n</hash>\n'

Is it possible to reuse OAuth1 authorization tokens?

Answer

I was misunderstanding the whole process. We do not need to save the request_token, request_token_secret and pin, but the access_token and the access_token_secret.

The process is actually:

  1. Use request_token, request_token_secret and pin to get access_token and access_token_secret
  2. Save access_token and access_token_secret (to database, or whatever)
  3. Next time, reuse access_token and access_token_secret

This is my corrected test code:

from rauth.service import OAuth1Serviceclass TwitterClient:KNOWN_USERS = { # (access_token, access_token_secret)'user1' : ("xxx", "yyy")}def __init__(self):# Get a real consumer key & secret from https://dev.twitter.com/apps/newself.twitter = OAuth1Service(name='twitter',consumer_key=TWITTER_CONSUMER_KEY,consumer_secret=TWITTER_CONSUMER_SECRET,request_token_url='https://api.twitter.com/oauth/request_token',access_token_url='https://api.twitter.com/oauth/access_token',authorize_url='https://api.twitter.com/oauth/authorize',base_url='https://api.twitter.com/1/')def new_session(self):request_token, request_token_secret = self.twitter.get_request_token()authorize_url = self.twitter.get_authorize_url(request_token)print 'Visit this URL in your browser: ' + authorize_urlpin = raw_input('Enter PIN from browser: ')session = self.twitter.get_auth_session(request_token,request_token_secret,method='POST',data={'oauth_verifier': pin})print session.access_token, session.access_token_secret # Save this to databasereturn sessiondef reuse_session(self, user):access_token, access_token_secret = self.KNOWN_USERS[user]session = self.twitter.get_session((access_token, access_token_secret))return sessiondef init_session(self, user):if user in self.KNOWN_USERS : session = self.reuse_session(user)else                        : session = self.new_session()return sessiondef list_tweets(self, user):session = self.init_session(user)params = {'include_rts': 1,  # Include retweets'count': 10}       # 10 tweetsr = session.get('statuses/home_timeline.json', params=params)for i, tweet in enumerate(r.json(), 1):handle = tweet['user']['screen_name'].encode('utf-8')text = tweet['text'].encode('utf-8')print '{0}. @{1} - {2}'.format(i, handle, text)tc = TwitterClient()tc.list_tweets('user1')
https://en.xdnf.cn/q/73403.html

Related Q&A

Cant run bjam to compile boost python tutorial

I am trying to follow this tutorial on wrapping C++ code for python for Windows.I installed python.Downloaded the latest version of boost(1_55).First I ran bootstrap.bat to build bjam.exe.Next, I confi…

How to use np.empty inside numba compiled function; Error message All templates rejected

I ran into this weird error when trying to use np.empty in a function definition compiled with numba, and turning on nopython=True to make sure optimized typing is in effect. Its weird because numba cl…

Python ValueError: Invalid header name b:authority

I see the : is error, but I cant find a way to solve it. ValueError: Invalid header name b:authorityIts the error:File "tmall.py", line 23, in get_url response = sessions.get(url=url,headers…

Psycopg2 callproc and sql parameters

I got some SQL functionCREATE OR REPLACE FUNCTION tools.update_company(IN company_id integer, OUT value integer)RETURNS integer AS$BODY$BEGIN select * into value from function_making_int(company_id) E…

How to write Huffman coding to a file using Python?

I created a Python script to compress text by using the Huffman algorithm. Say I have the following string:string = The quick brown fox jumps over the lazy dogRunning my algorithm returns the following…

How to know if a Python multiprocessing.Lock is released or not?

>>> l = Lock() >>> l.acquire() True >>> l.release() >>> l.release() Traceback (most recent call last):File "<stdin>", line 1, in <module> Value…

How do you ensure a Celery chord callback gets called with failed subtasks?

I am using a Chord in Celery to have a callback that gets called when a Group of parallel tasks finish executing. Specifically, I have a group of functions that wrap calls to an external API. I want to…

Unpacking nested C structs in Python

I am trying to unpack a C struct that is handed to my Python program in binary form and includes another nested struct. The relevant part of the C header looks like this:typedef struct {uint8_t seq;uin…

Remove black borders on images with watermarks in Python

I have a bunch of image I would like to uniformise by removing black borders. Usually I use the Trim function of Imagemagick with the fuzz parameters but in the case the image have some watermark the r…

scipy cdist with sparse matrices

I need to calculate the distances between two sets of vectors, source_matrix and target_matrix.I have the following line, when both source_matrix and target_matrix are of type scipy.sparse.csr.csr_matr…