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:
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:
InsightProse 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 issue starts with limits to posting Tweets on behalf of customers that have been severely reduced for the free access variant of the API 23:
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.
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.
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:
@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.
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)}")
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).
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:
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)
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.
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 ↩
Haftungsausschluss: Alle bereitgestellten Ressourcen stammen teilweise aus dem Internet. Wenn eine Verletzung Ihres Urheberrechts oder anderer Rechte und Interessen vorliegt, erläutern Sie bitte die detaillierten Gründe und legen Sie einen Nachweis des Urheberrechts oder Ihrer Rechte und Interessen vor und senden Sie ihn dann an die E-Mail-Adresse: [email protected] Wir werden die Angelegenheit so schnell wie möglich für Sie erledigen.
Copyright© 2022 湘ICP备2022001581号-3