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:
In case youâre interested in a quick run-down of what InsightProse actualy is, keep reading on. Alternatively, you can skip straight to Twitter / X API chapter to get straight to the topic of discussion.
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:
All of this content takes into account:
Twitter - X API v1.1 API deprecation controversy The moment of v1.1 deprecationInsightProse helps you promote your original content such that you can focus on long form content writing.
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 pricingThe 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:
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 XThis 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:
v2.0 API ImplementationThe 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.
The basic OAuth2.0 implementation flow5 for X is demonstrated in the following diagram:
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:
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;
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:
python@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 jobSince 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
lifespanevent 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:
python@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)
pythonasync 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)}")
OAuth2.0 offers several benefits over v1.0a:
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.
To solve this, you need to ensure:
This is how you revoken the access_token:
pythonasync 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:
pythonasync 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)
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.
ConclusionUnfortunately 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!
Thanks for reading.
RetroSearch is an open source project built by @garambo | Open a GitHub Issue
Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo
HTML:
3.2
| Encoding:
UTF-8
| Version:
0.7.4