如何在Django中用Tweepy实现用户认证并保存访问令牌及密钥
解决Tweepy Web应用中多用户授权与Token存储问题
嘿,我完全懂你的困惑——自己用固定Token调用Twitter API没问题,但要做Web应用让其他用户授权、还得存他们的Token来监控账号,这中间的流程确实对新手不太友好。我给你整理了一套完整的可行方案,带代码示例,一步步来:
核心原理:Twitter OAuth 1.0a 授权流程
你之前用的是直接使用自己的Token的方式,而Web应用需要走「用户授权流程」:
- 你的应用向Twitter请求一个临时的请求Token
- 用户跳转到Twitter的授权页面,允许你的应用访问其账号
- 用户授权后跳转回你的应用,你用返回的验证码换取用户的永久Access Token和Secret
- 存储这些Token,之后就能用它们代表用户调用API
完整代码示例(用Flask做轻量Web应用)
我用Flask来演示,因为它简单易上手,适合新手理解:
1. 先安装依赖
pip install tweepy flask
2. 完整应用代码
from flask import Flask, request, redirect, url_for, session import tweepy import sqlite3 # 替换成你的开发者信息 CONSUMER_KEY = "你的Consumer Key" CONSUMER_SECRET = "你的Consumer Secret" CALLBACK_URL = "http://localhost:5000/callback" # 要和Twitter开发者后台的回调URL一致 app = Flask(__name__) # 初始化数据库(第一次运行会自动创建) def init_db(): conn = sqlite3.connect('twitter_users.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS users (user_id TEXT PRIMARY KEY, screen_name TEXT, access_token TEXT, access_token_secret TEXT)''') conn.commit() conn.close() init_db() @app.route('/') def home(): # 1. 创建OAuthHandler,设置回调URL auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET, CALLBACK_URL) # 2. 获取请求Token和授权URL try: redirect_url = auth.get_authorization_url() # 临时保存request_token和secret(这里用session,生产环境可以用更安全的存储) session['request_token'] = auth.request_token['oauth_token'] session['request_token_secret'] = auth.request_token['oauth_token_secret'] return redirect(redirect_url) except tweepy.TweepyException as e: return f"获取授权URL失败: {str(e)}" @app.route('/callback') def callback(): # 3. 从回调参数中获取oauth_verifier oauth_verifier = request.args.get('oauth_verifier') request_token = session.get('request_token') request_token_secret = session.get('request_token_secret') if not oauth_verifier or not request_token or not request_token_secret: return "授权失败:缺少必要参数" # 4. 用请求Token和verifier换取用户的Access Token auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET) auth.request_token = {'oauth_token': request_token, 'oauth_token_secret': request_token_secret} try: access_token, access_token_secret = auth.get_access_token(oauth_verifier) # 5. 获取用户信息 api = tweepy.API(auth) user = api.me() # 6. 存储用户Token到数据库 conn = sqlite3.connect('twitter_users.db') c = conn.cursor() c.execute('REPLACE INTO users (user_id, screen_name, access_token, access_token_secret) VALUES (?,?,?,?)', (user.id_str, user.screen_name, access_token, access_token_secret)) conn.commit() conn.close() return f"授权成功!已保存用户@{user.screen_name}的Token" except tweepy.TweepyException as e: return f"换取Access Token失败: {str(e)}" @app.route('/monitor/<user_id>') def monitor_user(user_id): # 7. 从数据库取出Token,访问用户账号 conn = sqlite3.connect('twitter_users.db') c = conn.cursor() c.execute('SELECT access_token, access_token_secret FROM users WHERE user_id = ?', (user_id,)) result = c.fetchone() conn.close() if not result: return "找不到该用户的Token" access_token, access_token_secret = result auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET) auth.set_access_token(access_token, access_token_secret) api = tweepy.API(auth) # 示例:获取用户最新5条推文 try: tweets = api.user_timeline(count=5) tweet_texts = [f"- {tweet.text}" for tweet in tweets] return "<h3>用户最新推文:</h3>" + "<br>".join(tweet_texts) except tweepy.TweepyException as e: return f"访问用户账号失败: {str(e)}" if __name__ == '__main__': app.secret_key = '你的随机密钥' # 生产环境要换成安全的密钥 app.run(debug=True)
关键注意事项
- Twitter开发者后台配置:一定要在你的项目里把「Callback URL」设置成和代码里的
CALLBACK_URL一致(比如http://localhost:5000/callback),否则授权会失败。 - Token存储安全:示例里用了SQLite,生产环境建议用更安全的数据库(比如PostgreSQL),并且要加密存储
access_token_secret,避免泄露。 - 权限申请:在Twitter开发者后台,确保你的应用申请了足够的权限(比如「Read」权限用于监控用户推文),不然调用API会返回权限错误。
- Session存储:示例里用Flask的session存临时请求Token,生产环境可以用Redis等分布式存储,避免服务器重启后丢失数据。
测试流程
- 替换代码里的开发者信息和密钥
- 运行应用,访问
http://localhost:5000 - 跳转到Twitter授权页面,登录其他用户账号并授权
- 授权成功后,会显示保存Token的提示
- 访问
http://localhost:5000/monitor/用户ID(用户ID可以从数据库里查看),就能看到该用户的最新推文了
这样你就能实现多用户授权,并且保存他们的Token来监控账号啦!
内容的提问来源于stack exchange,提问作者Piyush
相关产品推荐
相关产品推荐

