How to implement the Twitter - X API using Python FastAPI

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
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:

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.
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.
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 ¹.

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 ² ³:

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 ⁴

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 flow⁵ for X is demonstrated in the following diagram:

Login with X button on your website.
Login is redirected to your API service.
Your API; a. forwards X Authorization request by; b. redirecting the user to the X website, c. to request scope Authorization.
The user has Authorized your application to access the requested scope of information and access to X user account details.
Request authorization ‘code’ forwarded to your API ‘/auth/x’ service.
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.
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:

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 settings⁶ 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 apscheduler⁷ 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 event⁸, 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 community⁹

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:

revoke the current active access_token within its 2 hours validity window,
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 dashboard¹⁰.

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

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.

https://x.com/XDevelopers/status/1649191520250245121 - Twitter API v1.1 deprecation announcement ↩
https://developer.x.com/en/docs/x-api/getting-started/about-x-api Twitter API Access Levels and versions ↩
https://developer.x.com/en/docs/x-api/rate-limits#v2-limits-free Twitter API v2.0 Rate limits for Free Access ↩
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) ↩
https://developer.x.com/en/docs/authentication/oauth-2-0/user-access-token Twitter OAuth2.0 Authorization Code Flow with PKCE ↩
https://developer.twitter.com/en/portal/dashboard Twitter Developer Dashboard ↩
https://apscheduler.readthedocs.io/en/3.x/# Advanced Python Scheduler ↩
https://fastapi.tiangolo.com/advanced/events/ FastAPI lifespan events ↩
https://devcommunity.x.com/t/access-token-expires-in/164425/13 Access token expires_in and Refresh token expiry ↩
https://developer.twitter.com/en/portal/products/free Twitter Developer Dashboard - Your API endpoint access overview ↩

Blog