”工欲善其事,必先利其器。“—孔子《论语.录灵公》
首页 > 编程 > 如何使用 Python FastAPI 实现 Twitter - X API

如何使用 Python FastAPI 实现 Twitter - X API

发布于2024-11-01
浏览:379

Introduction

While building my latest app, insightprose.com, I’ve gathered quite a few learnings in its 3 month development cycle. I wanted to start with Twitter integration since it’s a key component in the application feature-set of InsightProse.

I’ll be discussing:

  • the current state of X API v2.0,
  • where the Twitter API came from (version 1.1) briefly,
  • and how to implement OAuth2 auth flow with X API version 2.

In case you’re interested in a quick run-down of what InsightProse actualy is, keep reading on. Alternatively:

  • Twitter - X API
    • v1.1 API deprecation controversy
    • v2.0 API Implementation
    • Common questions and answers
  • Conclusion

What is InsightProse?

InsightProse is a social media and SEO content generator, using your original long form article, to distill one or more concepts into short articles.

These short articles are called Insights, these can then be used to create:

  1. Twitter threads, generates a Twitter thread of maximum 3 tweets to enhance social media reach, and consequently improve readership of the original article. These threads can be scheduled up to 1 month in advance.
  2. LinkedIn Articles, generates a short article of maximum 3000 characters that explains, in your branding style, a concept from the original article with the intention of getting people to read that original long form article.
  3. Blog posts; can be placed on a blog or website in order to enhance SEO, improve keyword relevance, and act as more concise Questions and Answers article that references and mirrors the original article. This format includes original code blocks (if any) and images where relevant for optimal concept explanation.

All of this content takes into account:

  • Your brand profile:. In other words, how you formulate text, how you explain ideas, and the target audience. This is to ensure the generated content does not read like AI generated content, but it’s like you’ve written it.
  • Keywords from the original article: In order to ensure relevance and SEO improvement, keywords from the original article are used appropriately and repeatedly in case you want to repost the insight prose on your blog for SEO purposes.
  • It takes into account “Questions and Answers” format that is used by many to search for answers using Google or other search engines. Again, with the purpose to improve relevance for those specific sets of keywords.

InsightProse helps you promote your original content such that you can focus on long form content writing.


Twitter - X API

v1.1 API deprecation controversy

The moment of v1.1 deprecation

Twitter / X used to be generous with its API usage towards developers, likely because they had income from advertisers primarily and no subscription services.

Now, this model has changed to a subscription oriented model with most advertisers dropping off. That has also affected the “user friendliness” towards developers. In a very negative way.

It started with the official announcement back in April 2023 the v1.1 API was being deprecated 1.

Today, we are deprecating our Premium v1.1 API, including Premium Search and Account Activity API.

You’ll notice in the thread of this announcement that there’s no love for this change, and that’s because the fees are not reasonable whatsoever.

The new rate limits and pricing

The issue starts with limits to posting Tweets on behalf of customers that have been severely reduced for the free access variant of the API 2 3:

  • v1.1: 300 posts per 3 hours = 2400 posts per 24 hours.
  • v2.0: 50 posts per 24 hours.

This is a factor of 48(!) reduction in Tweet post allowance. In order to mitigate some of these rate limiting issues, you can upgrade to the “Basic” X API.

1667 Tweets per 24 hours costing 100USD/month 4

You can imagine if you’re running a small SAAS product. In this case, 100 USD, is double the price of my infrastructure running cost on Digital Ocean. Double!

To further make the point, my infrastructure is a Kubernetes 2 Node cluster. For many developers that use Firebase and a free static site hosting solution such as AWS S3 or Cloudflare pages. They will pay near 0 USD per month to get bootstrapped.

The consequences, and my recommendation to X

This pricing means that posting Tweets on behalf of your customer needs to be severely capped or put on higher pricing tiers to get to the sufficient revenue to make it sustainable to pay 1200 USD / year to X.

I’m hoping that X will revise its pricing to considering smaller SAAS products and companies use-cases and enable them to integrate with X at reasonable prices.

I would recommend the following subscription tier to be added:

The introduction of a “Startup” tier 20 USD/month subscription would cater to starting business owners that want to built a quality service around the X eco-system. The current “Basic” 100 USD/month subscription, and “Free” options, are too expensive and too restrictive respectively.

v2.0 API Implementation

The basic OAuth2.0 implementation flow5 for X is demonstrated in the following diagram:

How to implement the Twitter - X API using Python FastAPI

  1. Login with X button on your website.
  2. Login is redirected to your API service.
  3. Your API; a. forwards X Authorization request by; b. redirecting the user to the X website, c. to request scope Authorization.
  4. The user has Authorized your application to access the requested scope of information and access to X user account details.
  5. Request authorization ‘code’ forwarded to your API ‘/auth/x’ service.
  6. Your auth X service receives the user data and X details such as username and refresh and access tokens through the coded response from the X API.
  7. Return your application refresh and access tokens to signal authorization is completed, user is logged in.

API /login/x (Step 2, 3, and 4)

The function of this API is to forward the user request to X such that they can authorize your APP to access the user’s account data and to post on behalf of this user.

In your FastAPI implementation you probably have a centralized api.py file that you add all individual API routes to:

api_router = APIRouter()
api_router.include_router(login.router, tags=["login"])

In the ./endpoints/login.py I would have the following route:

@router.get('/login/x')
async def login_twitter(request: Request):
    """Handle Twitter login using redirect to Twitter."""
    return await twitter.initiate_twitter_login(request)

Then to create the redirect url, we would do the following within the initiate_twitter_login function:

  • generate_oauth_params: create the required code verifier, challenge and state parameters.
  • create_authorize_url: are used to create the full redirect url to X API, including the scope. The minimum scope to post Tweets on behalf of your users: tweet.read, users.read, tweet.write. The offline.access scope is to get a refresh_token from X.
import secrets
import base64
import hashlib
from fastapi.responses import RedirectResponse

def _generate_oauth_params():
    code_verifier = secrets.token_urlsafe(32)
    code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode('utf-8')).digest()).decode('utf-8').rstrip('=')
    state = secrets.token_urlsafe(32)
    return code_verifier, code_challenge, state

def _create_authorize_url(code_challenge: str, state: str) -> str:
    params = {
        'response_type': 'code',
        'client_id': settings.TWITTER_CLIENT_ID,
        'redirect_uri': settings.TWITTER_CALLBACK_URL,
        'state': state,
        'code_challenge': code_challenge,
        'code_challenge_method': 'S256',
        'scope': 'tweet.read users.read tweet.write offline.access'
    }
    return f"https://x.com/i/oauth2/authorize?{urlencode(params)}"

async def initiate_twitter_login(request: Request):
    code_verifier, code_challenge, state = _generate_oauth_params()
    return RedirectResponse(_create_authorize_url(code_challenge, state))

This redirect will present you with a request screen from X;

  • initiated by step 3,
  • in step 4 the user authorizes the app,
  • which leads to the request send by X as step 5:

How to implement the Twitter - X API using Python FastAPI

API /auth/x (Steps 5, 6 and 7)

This endpoint receives the authorization from X, this is why you need to configure the callback API in the X settings6 such that X knows where to forward this API call to:

@router.get('/auth/twitter')
async def auth_twitter(request: Request, db: Session = Depends(deps.get_db)):
    access_token, refresh_token = await twitter.handle_twitter_callback(request, db)

    return RedirectResponse(url=f"{settings.FRONTEND_URL}/app/auth/callback?access_token={access_token}&refresh_token={refresh_token}")

This API takes care of validation of the OAuth state secret that we created in the first API, which should match here.

Hence, we start with a state check to ensure that the request was initiated from this session and not from somewhere else.

The code contains the X tokens, because we requested the offline.access scope we also get a refresh token next to the regular access token.

async def handle_twitter_callback(request: Request, db: Session): Tuple
    if request.query_params.get('state') != request.session.get('oauth_state'):
        raise HTTPException(status_code=400, detail="Invalid state parameter")

    code = request.query_params.get('code')
    if not code:
        raise HTTPException(status_code=400, detail="No authorization code provided")

    token_data = await _exchange_code_for_token(code, request.session.get('code_verifier'))
    twitter_user_info = await get_twitter_user_info(token_data['access_token'])

    # Create your Application user with X details here

    request.session.pop('code_verifier', None)
    request.session.pop('oauth_state', None)

    # access_token, refresh_token
    return access_token, new_refresh_token

To receive this; we execute token_data = await _exchange_code_for_token(code, request.session.get('code_verifier'))

async def _exchange_code_for_token(code: str, code_verifier: str) -> dict:
    url = 'https://api.x.com/2/oauth2/token'
    data = {
        'code': code,
        'grant_type': 'authorization_code',
        'client_id': settings.TWITTER_CLIENT_ID,
        'redirect_uri': settings.TWITTER_CALLBACK_URL,
        'code_verifier': code_verifier
    }
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
    }
    auth = httpx.BasicAuth(settings.TWITTER_CLIENT_ID, settings.TWITTER_CLIENT_SECRET)
    return await _make_twitter_api_call('POST', url, headers=headers, data=data, auth=auth)

With the received access token, we can pull in the user data using get_twitter_user_info(token_data['access_token']) function:

async def _get_twitter_user_info(access_token: str) -> dict:
    url = 'https://api.twitter.com/2/users/me'
    params = {
        'user.fields': 'id,name,username,profile_image_url'
    }
    headers = {
        'Authorization': f"Bearer {access_token}"
    }
    return await _make_twitter_api_call('GET', url, headers=headers, params=params)

Once you have the Twitter / X user data, you create your application user profile with your application access credentials and return that to your user and they’re logged in.

Refresh token cron job

Since OAuth2.0 has limited access token validity time, 7200 seconds in case of Twitter / X OAuth2.0. We need to manage automatic renewal of this token in the background.

I recommend using a task scheduling system or a cron job that automatically checks your user table or token issuance table for expired / about to be expired access tokens.

In my application I’m using apscheduler7 to schedule tasks on the same application server as the API.

apscheduler is a low profile scheduling library that will “attach” itself to the FastAPI application lifespan event hook.

If you want to know more about how to use it, let me know on social media!

This is configured as a lifespan event8, that way it will be running in the background from the moment your FastAPI server is online. This lifespan event uses an async context manager to handle the two events; startup, and shutdown (after the yield) to start and stop the scheduler:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Setup code (runs before the app starts)
    try:
        wait_for_db()
        scheduler = create_scheduler()
        setup_scheduler(scheduler)
        scheduler.start()
        logger.info("Application startup completed successfully")
    except Exception as e:
        logger.error(f"Error during application startup: {e}")
        raise
    yield
    # Cleanup code (runs when the app is shutting down)
    scheduler.shutdown()
    logger.info("Application shutdown")

app = FastAPI(
    lifespan=lifespan,
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.OPENAPI_URL}"
)

In the task that is defined, we want to run this regularly to refresh expired Twitter / X access tokens:

Be aware that you need to revoke your access tokens in case you’re refreshing them within the 2 hour lifespan to avoid token refresh failure errors (see Why am I getting refresh token failure with Twitter / X API)

async def refresh_all_expired_tokens(db: Session):
    now = datetime.now(timezone.utc)
    users_with_expired_tokens = user_crud.get_users_with_expired_tokens(db, now)

    for user in users_with_expired_tokens:
        try:
            new_token = await refresh_twitter_token_by_user_id(user.id, db)
            if new_token:
                logger.info(f"Successfully refreshed token for user {user.id}")
            else:
                logger.warning(f"Failed to refresh token for user {user.id}")
        except Exception as e:
            logger.error(f"Error refreshing token for user {user.id}: {str(e)}")

Common questions and answers

Why should you use OAuth 2.0 and not Oauth1.0a for Twitter / X API authorization?

OAuth2.0 offers several benefits over v1.0a:

  • Granular permission management vs. fine-grained:
  • Refresh tokens vs. no expiry access tokens.
  • Eventual v1.0.a deprecation, requiring migration effort to OAuth2.0

Generally, security and access controls have improved in version 2.0. However, that does mean you have to manage access token validity in your application automatically (see Refresh token cron job).

What is the validity of the refresh token and access tokens for Twitter / X API?

There’s no clear documentation on the official developer.x.com that clarifies expiration of the refresh_token provided, but apparently it’s 6 months according to one user in the x community9

Access token has a return value with expires_in set to 7200 seconds, which is 2 hours.

Why am I getting refresh token failure with Twitter / X API

To solve this, you need to ensure:

  1. revoke the current active access_token within its 2 hours validity window,
  2. after the 2 hour window, you can use the refresh_token to renew the access_token without revoking the access_token.

This is how you revoken the access_token:

async def revoke_twitter_token(token: str) -> bool:
    url = 'https://api.x.com/2/oauth2/revoke'
    data = {
        'token': token,
        'client_id': settings.TWITTER_CLIENT_ID
    }
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    auth = httpx.BasicAuth(settings.TWITTER_CLIENT_ID, settings.TWITTER_CLIENT_SECRET)
    return await _make_twitter_api_call('POST', url, headers=headers, data=data, auth=auth)

This is how you refresh the access_token:

async def refresh_twitter_token(refresh_token: str) -> dict:
    url = 'https://api.twitter.com/2/oauth2/token'
    data = {
        'refresh_token': refresh_token,
        'grant_type': 'refresh_token',
        'client_id': settings.TWITTER_CLIENT_ID,
    }
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    # Prepare Basic Auth
    auth = httpx.BasicAuth(settings.TWITTER_CLIENT_ID, settings.TWITTER_CLIENT_SECRET)
    return await _make_twitter_api_call('POST', url, headers=headers, data=data, auth=auth)

Which API's do I have access to in the Free version of X API v2.0?

You can see a full listing of your API access when you register an account, in your development dashboard10.

Or you can go to the About X API page that is publicly accessible2.


Conclusion

Unfortunately with the advent of Twitter / X API version 2, the usability for small products and applications has been severely diminished with a high ticket entrance fee of 100 USD per month for the Basic plan. For the Free version, a very limited allowance of 50 Tweets per day.

There’s been backlash from day one when this new business model was announced, however we haven’t seen X make any moves to amend or improve their API access for smaller startups and businesses with low revenue.

Luckily, the implementation of the X API is pretty straightforward as I’ve hopefully demonstrated in this article. There are some caveats when it comes to access token / refresh token issues that have been reported online very frequently. But with a proper implementation of access token revoke, before a refresh this should be resolved.

Have you encountered any problems implementing the new X API v2.0? If so, lets discuss below!

Thanks for reading.


  1. https://x.com/XDevelopers/status/1649191520250245121 - Twitter API v1.1 deprecation announcement ↩

  2. https://developer.x.com/en/docs/x-api/getting-started/about-x-api Twitter API Access Levels and versions ↩

  3. https://developer.x.com/en/docs/x-api/rate-limits#v2-limits-free Twitter API v2.0 Rate limits for Free Access ↩

  4. https://developer.x.com/en/docs/x-api/rate-limits#v2-limits-basic Twitter API v2.0 Rate limits for Basic Paid Access (100USD / month) ↩

  5. https://developer.x.com/en/docs/authentication/oauth-2-0/user-access-token Twitter OAuth2.0 Authorization Code Flow with PKCE ↩

  6. https://developer.twitter.com/en/portal/dashboard Twitter Developer Dashboard ↩

  7. https://apscheduler.readthedocs.io/en/3.x/# Advanced Python Scheduler ↩

  8. https://fastapi.tiangolo.com/advanced/events/ FastAPI lifespan events ↩

  9. https://devcommunity.x.com/t/access-token-expires-in/164425/13 Access token expires_in and Refresh token expiry ↩

  10. https://developer.twitter.com/en/portal/products/free Twitter Developer Dashboard - Your API endpoint access overview ↩

版本声明 本文转载于:https://dev.to/rolfstreefkerk/how-to-implement-the-twitter-x-api-using-python-fastapi-55da?1如有侵犯,请联系[email protected]删除
最新教程 更多>
  • 如何使用Depimal.parse()中的指数表示法中的数字?
    如何使用Depimal.parse()中的指数表示法中的数字?
    在尝试使用Decimal.parse(“ 1.2345e-02”中的指数符号表示法表示的字符串时,您可能会遇到错误。这是因为默认解析方法无法识别指数符号。 成功解析这样的字符串,您需要明确指定它代表浮点数。您可以使用numbersTyles.Float样式进行此操作,如下所示:[&& && && ...
    编程 发布于2025-05-06
  • 切换到MySQLi后CodeIgniter连接MySQL数据库失败原因
    切换到MySQLi后CodeIgniter连接MySQL数据库失败原因
    无法连接到mySQL数据库:故障排除错误消息要调试问题,建议将以下代码添加到文件的末尾.//config/database.php并查看输出: ... ... 回声'... echo '<pre>'; print_r($db['default']); echo '</pr...
    编程 发布于2025-05-06
  • 如何使用Python有效地以相反顺序读取大型文件?
    如何使用Python有效地以相反顺序读取大型文件?
    在python 反向行读取器生成器 == ord('\ n'): 缓冲区=缓冲区[:-1] 剩余_size- = buf_size lines = buffer.split('\ n'....
    编程 发布于2025-05-06
  • 如何解决AppEngine中“无法猜测文件类型,使用application/octet-stream...”错误?
    如何解决AppEngine中“无法猜测文件类型,使用application/octet-stream...”错误?
    appEngine静态文件mime type override ,静态文件处理程序有时可以覆盖正确的mime类型,在错误消息中导致错误消息:“无法猜测mimeType for for file for file for [File]。 application/application/octet...
    编程 发布于2025-05-06
  • 在Pandas中如何将年份和季度列合并为一个周期列?
    在Pandas中如何将年份和季度列合并为一个周期列?
    pandas data frame thing commans date lay neal and pree pree'和pree pree pree”,季度 2000 q2 这个目标是通过组合“年度”和“季度”列来创建一个新列,以获取以下结果: [python中的concate...
    编程 发布于2025-05-06
  • 左连接为何在右表WHERE子句过滤时像内连接?
    左连接为何在右表WHERE子句过滤时像内连接?
    左JOIN CONUNDRUM:WITCHING小时在数据库Wizard的领域中变成内在的加入很有趣,当将c.foobar条件放置在上面的Where子句中时,据说左联接似乎会转换为内部连接。仅当满足A.Foo和C.Foobar标准时,才会返回结果。为什么要变形?关键在于其中的子句。当左联接的右侧值...
    编程 发布于2025-05-06
  • CSS强类型语言解析
    CSS强类型语言解析
    您可以通过其强度或弱输入的方式对编程语言进行分类的方式之一。在这里,“键入”意味着是否在编译时已知变量。一个例子是一个场景,将整数(1)添加到包含整数(“ 1”)的字符串: result = 1 "1";包含整数的字符串可能是由带有许多运动部件的复杂逻辑套件无意间生成的。它也可以是故意从单个真理...
    编程 发布于2025-05-06
  • Java为何无法创建泛型数组?
    Java为何无法创建泛型数组?
    通用阵列创建错误 arrayList [2]; JAVA报告了“通用数组创建”错误。为什么不允许这样做?答案:Create an Auxiliary Class:public static ArrayList<myObject>[] a = new ArrayList<myO...
    编程 发布于2025-05-06
  • 如何克服PHP的功能重新定义限制?
    如何克服PHP的功能重新定义限制?
    克服PHP的函数重新定义限制在PHP中,多次定义一个相同名称的函数是一个no-no。尝试这样做,如提供的代码段所示,将导致可怕的“不能重新列出”错误。 但是,PHP工具腰带中有一个隐藏的宝石:runkit扩展。它使您能够灵活地重新定义函数。 runkit_function_renction_re...
    编程 发布于2025-05-06
  • 如何修复\“常规错误:2006 MySQL Server在插入数据时已经消失\”?
    如何修复\“常规错误:2006 MySQL Server在插入数据时已经消失\”?
    How to Resolve "General error: 2006 MySQL server has gone away" While Inserting RecordsIntroduction:Inserting data into a MySQL database can...
    编程 发布于2025-05-06
  • 如何使用替换指令在GO MOD中解析模块路径差异?
    如何使用替换指令在GO MOD中解析模块路径差异?
    在使用GO MOD时,在GO MOD 中克服模块路径差异时,可能会遇到冲突,其中可能会遇到一个冲突,其中3派对软件包将另一个带有导入套件的path package the Imptioned package the Imptioned package the Imported tocted pac...
    编程 发布于2025-05-06
  • Java字符串非空且非null的有效检查方法
    Java字符串非空且非null的有效检查方法
    检查字符串是否不是null而不是空的if (str != null && !str.isEmpty())Option 2: str.length() == 0For Java versions prior to 1.6, str.length() == 0 can be二手: if(str!= n...
    编程 发布于2025-05-06
  • PHP未来:适应与创新
    PHP未来:适应与创新
    PHP的未来将通过适应新技术趋势和引入创新特性来实现:1)适应云计算、容器化和微服务架构,支持Docker和Kubernetes;2)引入JIT编译器和枚举类型,提升性能和数据处理效率;3)持续优化性能和推广最佳实践。 引言在编程世界中,PHP一直是网页开发的中流砥柱。作为一个从1994年就开始发展...
    编程 发布于2025-05-06
  • MySQL中如何高效地根据两个条件INSERT或UPDATE行?
    MySQL中如何高效地根据两个条件INSERT或UPDATE行?
    在两个条件下插入或更新或更新 solution:的答案在于mysql的插入中...在重复键更新语法上。如果不存在匹配行或更新现有行,则此功能强大的功能可以通过插入新行来进行有效的数据操作。如果违反了唯一的密钥约束。实现所需的行为,该表必须具有唯一的键定义(在这种情况下为'名称'...
    编程 发布于2025-05-06
  • 如何简化PHP中的JSON解析以获取多维阵列?
    如何简化PHP中的JSON解析以获取多维阵列?
    php 试图在PHP中解析JSON数据的JSON可能具有挑战性,尤其是在处理多维数组时。 To simplify the process, it's recommended to parse the JSON as an array rather than an object.To do...
    编程 发布于2025-05-06

免责声明: 提供的所有资源部分来自互联网,如果有侵犯您的版权或其他权益,请说明详细缘由并提供版权或权益证明然后发到邮箱:[email protected] 我们会第一时间内为您处理。

Copyright© 2022 湘ICP备2022001581号-3